nexus-ai-api / app /core /database.py
github-actions[bot]
Auto Deploy from Monorepo: Merge pull request #72 from human13th2team/dev
df5e016
Raw History Blame Contribute Delete
945 Bytes
import os
from dotenv import load_dotenv
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
# .env ํŒŒ์ผ ๋กœ๋“œ
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
raise ValueError("DATABASE_URL์ด ์„ค์ •๋˜์ง€ ์•Š์•˜์Šต๋‹ˆ๋‹ค. .env ํŒŒ์ผ์„ ํ™•์ธํ•ด์ฃผ์„ธ์š”.")
# ๋น„๋™๊ธฐ ์—”์ง„ ์ƒ์„ฑ
engine = create_async_engine(DATABASE_URL, echo=True, connect_args={"statement_cache_size": 0})
# ๋น„๋™๊ธฐ ์„ธ์…˜ ๋ฉ”์ด์ปค
AsyncSessionLocal = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
# Base ํด๋ž˜์Šค ์ •์˜ (SQLAlchemy 2.0 style)
class Base(DeclarativeBase):
pass
# DB ์„ธ์…˜ ์˜์กด์„ฑ ์ฃผ์ž…์„ ์œ„ํ•œ ์ œ๋„ˆ๋ ˆ์ดํ„ฐ
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()