Function get_db_connection

service/server/database.py:305–325

Get database connection. Supports both SQLite and PostgreSQL.

()

def get_db_connection():
    """Get database connection. Supports both SQLite and PostgreSQL."""
    if using_postgres():
        if psycopg is None:
            raise RuntimeError(
                "PostgreSQL support requires psycopg. Install service requirements first."
            )
        conn = psycopg.connect(DATABASE_URL, row_factory=dict_row)
        return DatabaseConnection(conn, "postgres")

    db_path = _SQLITE_DB_PATH
    os.makedirs(os.path.dirname(db_path), exist_ok=True)

    conn = sqlite3.connect(db_path, timeout=30.0)
    conn.row_factory = sqlite3.Row

    # Enable WAL mode for better concurrent access
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")

    return DatabaseConnection(conn, "sqlite")


def get_database_status() -> dict[str, Any]: