60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
外部スクリプトからDBコンテナに接続するサンプル
|
|
"""
|
|
import os
|
|
import psycopg2
|
|
from psycopg2.extras import DictCursor
|
|
|
|
# 環境変数から接続情報を取得
|
|
DB_CONFIG = {
|
|
'host': os.getenv('PG_HOST', 'localhost'),
|
|
'port': os.getenv('PG_PORT', '5432'),
|
|
'database': os.getenv('POSTGRES_DBNAME', 'rogdb'),
|
|
'user': os.getenv('POSTGRES_USER', 'admin'),
|
|
'password': os.getenv('POSTGRES_PASS', 'admin123456')
|
|
}
|
|
|
|
def connect_to_db():
|
|
"""データベースに接続"""
|
|
try:
|
|
conn = psycopg2.connect(**DB_CONFIG)
|
|
print(f"✅ データベースに接続成功: {DB_CONFIG['host']}:{DB_CONFIG['port']}")
|
|
return conn
|
|
except psycopg2.Error as e:
|
|
print(f"❌ データベース接続エラー: {e}")
|
|
return None
|
|
|
|
def test_connection():
|
|
"""接続テスト"""
|
|
conn = connect_to_db()
|
|
if conn:
|
|
try:
|
|
with conn.cursor(cursor_factory=DictCursor) as cur:
|
|
cur.execute("SELECT version();")
|
|
version = cur.fetchone()
|
|
print(f"PostgreSQL バージョン: {version[0]}")
|
|
|
|
# テーブル一覧を取得
|
|
cur.execute("""
|
|
SELECT tablename FROM pg_tables
|
|
WHERE schemaname = 'public'
|
|
ORDER BY tablename;
|
|
""")
|
|
tables = cur.fetchall()
|
|
print(f"テーブル数: {len(tables)}")
|
|
for table in tables[:5]: # 最初の5個を表示
|
|
print(f" - {table[0]}")
|
|
|
|
except psycopg2.Error as e:
|
|
print(f"❌ クエリ実行エラー: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
print("=== データベース接続テスト ===")
|
|
print(f"接続先: {DB_CONFIG['host']}:{DB_CONFIG['port']}")
|
|
print(f"データベース: {DB_CONFIG['database']}")
|
|
print(f"ユーザー: {DB_CONFIG['user']}")
|
|
test_connection()
|