chwellofficial/nt360Slides

1import os2from utils.get_env import get_app_data_directory_env, get_database_url_env3from urllib.parse import urlsplit, urlunsplit, parse_qsl4import ssl5 6 7def _ensure_sqlite_parent_dir(database_url: str) -> None:8    if not database_url.startswith("sqlite://"):9        return10 11    split_result = urlsplit(database_url)12    db_path = split_result.path13    if not db_path:14        return15 16    # sqlite URLs on Windows can start with /C:/..., normalize that for os.path.17    if os.name == "nt" and len(db_path) >= 3 and db_path[0] == "/" and db_path[2] == ":":18        db_path = db_path[1:]19 20    parent = os.path.dirname(db_path)21    if parent:22        os.makedirs(parent, exist_ok=True)23def _int_env(name: str, default: int) -> int:24    """Read an integer from an environment variable, falling back to *default*."""25    raw = os.getenv(name)26    if raw is None:27        return default28    try:29        return int(raw)30    except ValueError:31        return default32 33 34def get_pool_kwargs() -> dict:35    """Build SQLAlchemy engine pool keyword arguments from environment variables.36 37    Supported variables (all optional):38        DB_POOL_SIZE          – max persistent connections (default 5)39        DB_MAX_OVERFLOW       – extra connections above pool_size (default 10)40        DB_POOL_TIMEOUT       – seconds to wait for a connection (default 30)41        DB_POOL_RECYCLE       – seconds before a connection is recycled (default 1800)42        DB_POOL_PRE_PING      – enable connection liveness check (default true)43 44    For SQLite the pool settings are not applicable and an empty dict is45    returned, since SQLite uses ``StaticPool`` / ``NullPool`` by default.46    """47    return {48        "pool_size": _int_env("DB_POOL_SIZE", 5),49        "max_overflow": _int_env("DB_MAX_OVERFLOW", 10),50        "pool_timeout": _int_env("DB_POOL_TIMEOUT", 30),51        "pool_recycle": _int_env("DB_POOL_RECYCLE", 1800),52        "pool_pre_ping": os.getenv("DB_POOL_PRE_PING", "true").lower()53        not in ("false", "0", "no"),54    }55 56 57def get_database_url_and_connect_args() -> tuple[str, dict]:58    database_url = get_database_url_env() or "sqlite:///" + os.path.join(59        get_app_data_directory_env() or "/tmp/presenton", "fastapi.db"60    )61 62    _ensure_sqlite_parent_dir(database_url)63 64    if database_url.startswith("sqlite://"):65        database_url = database_url.replace("sqlite://", "sqlite+aiosqlite://", 1)66    elif database_url.startswith("postgresql://"):67        database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)68    elif database_url.startswith("mysql://"):69        database_url = database_url.replace("mysql://", "mysql+aiomysql://", 1)70    else:71        database_url = database_url72 73    connect_args = {}74    if "sqlite" in database_url:75        connect_args["check_same_thread"] = False76 77    try:78        split_result = urlsplit(database_url)79        if split_result.query:80            query_params = parse_qsl(split_result.query, keep_blank_values=True)81            driver_scheme = split_result.scheme82            for k, v in query_params:83                key_lower = k.lower()84                if key_lower == "sslmode" and "postgresql+asyncpg" in driver_scheme:85                    if v.lower() != "disable" and "sqlite" not in database_url:86                        connect_args["ssl"] = ssl.create_default_context()87 88            database_url = urlunsplit(89                (90                    split_result.scheme,91                    split_result.netloc,92                    split_result.path,93                    "",94                    split_result.fragment,95                )96            )97    except Exception:98        pass99 100    return database_url, connect_args101