55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
"""Flask extensions: per-request database connection and Gitea HTTP session.
|
|
|
|
These helpers depend on Flask's `g` and can only be used within app/request context.
|
|
"""
|
|
|
|
import logging
|
|
import sqlite3
|
|
from typing import Any
|
|
|
|
import requests
|
|
from flask import g
|
|
|
|
import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_db() -> sqlite3.Connection:
|
|
"""Return per-request SQLite connection with Row factory."""
|
|
if "db" not in g:
|
|
try:
|
|
conn = sqlite3.connect(config.KANBAN_DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
g.db = conn
|
|
except sqlite3.Error as exc:
|
|
logger.error("Failed to connect to kanban DB: %s", exc)
|
|
raise
|
|
return g.db
|
|
|
|
|
|
def close_db(exception: BaseException | None = None) -> None:
|
|
"""Teardown handler registered via app.teardown_appcontext."""
|
|
conn = g.pop("db", None)
|
|
if conn is not None:
|
|
conn.close()
|
|
|
|
|
|
def get_gitea_session() -> requests.Session:
|
|
"""Return per-request requests.Session for Gitea API calls."""
|
|
if "gitea_session" not in g:
|
|
session = requests.Session()
|
|
session.headers.update({"Accept": "application/json"})
|
|
|
|
# Try to load admin token
|
|
try:
|
|
with open(config.GITEA_TOKEN_PATH, "r", encoding="utf-8") as f:
|
|
token = f.read().strip()
|
|
if token:
|
|
session.headers["Authorization"] = f"token {token}"
|
|
except (FileNotFoundError, PermissionError, OSError) as exc:
|
|
logger.debug("Gitea token not available: %s", exc)
|
|
|
|
g.gitea_session = session
|
|
return g.gitea_session
|