| from collections.abc import Generator | |
| import os | |
| from pathlib import Path | |
| from urllib.parse import unquote, urlparse | |
| from sqlalchemy import create_engine | |
| from sqlalchemy.orm import Session, declarative_base, sessionmaker | |
| DEFAULT_DB_PATH = Path(__file__).resolve().parents[1] / "cronogramas_v2.db" | |
| DATABASE_URL = os.getenv("DATABASE_URL", f"sqlite:///{DEFAULT_DB_PATH}") | |
| connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {} | |
| engine = create_engine(DATABASE_URL, connect_args=connect_args) | |
| SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) | |
| Base = declarative_base() | |
| def sqlite_database_path() -> Path | None: | |
| if not DATABASE_URL.startswith("sqlite:///"): | |
| return None | |
| raw_path = DATABASE_URL.removeprefix("sqlite:///") | |
| if DATABASE_URL.startswith("sqlite:////"): | |
| parsed = urlparse(DATABASE_URL) | |
| raw_path = unquote(parsed.path) | |
| return Path(raw_path).expanduser().resolve() | |
| def get_db() -> Generator[Session, None, None]: | |
| db = SessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |