52 lines
1.1 KiB
Python
52 lines
1.1 KiB
Python
"""
|
|
config_db.py
|
|
데이터베이스 연결 설정 (PostgreSQL + asyncpg)
|
|
|
|
비동기 연결 풀 생성 및 관리.
|
|
"""
|
|
|
|
import asyncpg
|
|
from typing import Optional
|
|
from .config_system import (
|
|
DB_HOST,
|
|
DB_PORT,
|
|
DB_NAME,
|
|
DB_USER,
|
|
DB_PASSWORD,
|
|
DB_POOL_MIN,
|
|
DB_POOL_MAX,
|
|
)
|
|
|
|
# 글로벌 DB 풀 (앱 시작/종료 시 관리)
|
|
db_pool: Optional[asyncpg.Pool] = None
|
|
|
|
|
|
async def init_db_pool() -> asyncpg.Pool:
|
|
"""PostgreSQL 연결 풀 초기화"""
|
|
global db_pool
|
|
db_pool = await asyncpg.create_pool(
|
|
host=DB_HOST,
|
|
port=DB_PORT,
|
|
database=DB_NAME,
|
|
user=DB_USER,
|
|
password=DB_PASSWORD,
|
|
min_size=DB_POOL_MIN,
|
|
max_size=DB_POOL_MAX,
|
|
)
|
|
return db_pool
|
|
|
|
|
|
async def close_db_pool() -> None:
|
|
"""PostgreSQL 연결 풀 종료"""
|
|
global db_pool
|
|
if db_pool:
|
|
await db_pool.close()
|
|
db_pool = None
|
|
|
|
|
|
def get_db_pool() -> asyncpg.Pool:
|
|
"""현재 활성 DB 풀 반환. 없으면 RuntimeError"""
|
|
if not db_pool:
|
|
raise RuntimeError("DB pool not initialized. Call init_db_pool() first.")
|
|
return db_pool
|