mirror of
https://github.com/maxdorninger/MediaManager.git
synced 2026-04-18 01:54:00 +02:00
45fc771c16
String concatination can cause issues with special characters and can be a risk for URL parameter injection. fixes https://github.com/maxdorninger/MediaManager/issues/193
61 lines
1.4 KiB
Python
61 lines
1.4 KiB
Python
import logging
|
|
from contextvars import ContextVar
|
|
from typing import Annotated, Any, Generator
|
|
|
|
from fastapi import Depends
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine.url import URL
|
|
from sqlalchemy.orm import Session, declarative_base, sessionmaker
|
|
|
|
from media_manager.config import AllEncompassingConfig
|
|
|
|
log = logging.getLogger(__name__)
|
|
config = AllEncompassingConfig().database
|
|
|
|
db_url = URL.create(
|
|
"postgresql+psycopg",
|
|
config.user,
|
|
config.password,
|
|
config.host,
|
|
config.port,
|
|
config.dbname,
|
|
)
|
|
|
|
engine = create_engine(
|
|
db_url,
|
|
echo=False,
|
|
pool_size=10,
|
|
max_overflow=10,
|
|
pool_timeout=30,
|
|
pool_recycle=1800,
|
|
)
|
|
log.debug("initializing sqlalchemy declarative base")
|
|
Base = declarative_base()
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
def init_db() -> None:
|
|
log.debug("initializing database with following tables")
|
|
for table in Base.metadata.tables:
|
|
log.debug(f"Table: {table.title()}")
|
|
Base.metadata.create_all(engine)
|
|
|
|
|
|
def get_session() -> Generator[Session, Any, None]:
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
db.commit()
|
|
except Exception as e:
|
|
db.rollback()
|
|
log.critical(f"error occurred: {e}")
|
|
raise e
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
db_session: ContextVar[Session] = ContextVar("db_session")
|
|
|
|
|
|
DbSessionDependency = Annotated[Session, Depends(get_session)]
|