38 lines
1.0 KiB
Python
38 lines
1.0 KiB
Python
# backend/app/db.py
|
|
"""SQLAlchemy engine, session factory i Base declarativa."""
|
|
from contextlib import contextmanager
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
|
|
from typing import Generator
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
_engine = None
|
|
_SessionLocal = None
|
|
|
|
|
|
def init_db(database_url: str) -> None:
|
|
global _engine, _SessionLocal
|
|
connect_args = {"check_same_thread": False} if database_url.startswith("sqlite") else {}
|
|
_engine = create_engine(database_url, connect_args=connect_args, echo=False)
|
|
_SessionLocal = sessionmaker(bind=_engine, autocommit=False, autoflush=False)
|
|
Base.metadata.create_all(_engine)
|
|
|
|
|
|
@contextmanager
|
|
def get_session() -> Generator[Session, None, None]:
|
|
"""Context manager de sessió SQLAlchemy."""
|
|
if _SessionLocal is None:
|
|
raise RuntimeError("Database not initialized. Call init_db() first.")
|
|
db = _SessionLocal()
|
|
try:
|
|
yield db
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|