"""

SQLite connection management and user CRUD.

All application data lives in a single SQLite file (config.DATABASE_PATH).

Per-user encrypted data can use get_user_connection(user_id) when you attach

user-specific databases later; for now it returns the shared connection scoped

to the authenticated user context.

"""

from __future__ import annotations

import sqlite3

from contextlib import contextmanager

from pathlib import Path

from typing import Any, Generator, Iterator

import config

from database.models.users import (

CREATE_EMAIL_INDEX,

CREATE_USERNAME_INDEX,

USERS_TABLE_SQL,

UserRecord,

utc_now_iso,

)

from utils.encryption import encrypt_for_storage, generate_user_fernet_key

def _db_path() -> Path:

return Path(config.DATABASE_PATH)

def get_connection(*, row_factory: bool = True) -> sqlite3.Connection:

"""

Open (and initialize) the shared SQLite database.

row_factory=True enables dict-like row access via conn.row_factory.

"""

path = _db_path()

path.parent.mkdir(parents=True, exist_ok=True)

conn = sqlite3.connect(str(path), check_same_thread=False)

if row_factory:

conn.row_factory = sqlite3.Row

initialize_schema(conn)

return conn

@contextmanager

def db_session() -> Generator[sqlite3.Connection, None, None]:

"""Context manager that commits on success and rolls back on error."""

conn = get_connection()

try:

yield conn

conn.commit()

except Exception:

conn.rollback()

raise

finally:

conn.close()

def initialize_schema(conn: sqlite3.Connection) -> None:

"""Create tables and indexes if they do not exist."""

conn.executescript(USERS_TABLE_SQL)

conn.executescript(CREATE_USERNAME_INDEX)

conn.executescript(CREATE_EMAIL_INDEX)

conn.commit()

def get_user_connection(user_id: int) -> sqlite3.Connection:

"""

Return a database connection for operations scoped to a user.

Currently uses the shared app database; extend this to open per-user

SQLite files (e.g. data/users/{user_id}.db) when needed.

"""

conn = get_connection()

conn.execute("PRAGMA foreign_keys = ON")

# Store user context for optional query helpers

conn.execute("SELECT 1") # no-op; placeholder for future per-user DB split

conn._app_user_id = user_id # type: ignore[attr-defined]

return conn

# ---------------------------------------------------------------------------

# User CRUD

# ---------------------------------------------------------------------------

_USER_COLUMNS = (

"user_id, username, email, name, password_hash, "

"encryption_key, created_at, is_active"

)

def _row_to_user(row: sqlite3.Row | tuple[Any, ...]) -> UserRecord:

if isinstance(row, sqlite3.Row):

return UserRecord.from_row(tuple(row))

return UserRecord.from_row(row)

def get_user_by_id(user_id: int) -> UserRecord | None:

with db_session() as conn:

cur = conn.execute(

f"SELECT {_USER_COLUMNS} FROM users WHERE user_id = ? AND is_active = 1",

(user_id,),

)

row = cur.fetchone()

return _row_to_user(row) if row else None

def get_user_by_username(username: str) -> UserRecord | None:

with db_session() as conn:

cur = conn.execute(

f"SELECT {_USER_COLUMNS} FROM users WHERE username = ? AND is_active = 1",

(username.lower(),),

)

row = cur.fetchone()

return _row_to_user(row) if row else None

def get_user_by_email(email: str) -> UserRecord | None:

with db_session() as conn:

cur = conn.execute(

f"SELECT {_USER_COLUMNS} FROM users WHERE email = ? AND is_active = 1",

(email.lower(),),

)

row = cur.fetchone()

return _row_to_user(row) if row else None

def list_active_users() -> list[UserRecord]:

with db_session() as conn:

cur = conn.execute(

f"SELECT {_USER_COLUMNS} FROM users WHERE is_active = 1 ORDER BY username"

)

return [_row_to_user(r) for r in cur.fetchall()]

def create_user(

*,

username: str,

email: str,

name: str,

password_hash: str,

encryption_key_plain: str | None = None,

) -> UserRecord:

"""

Insert a new user. encryption_key is generated and stored encrypted unless

a plaintext key is supplied (e.g. migration).

"""

username = username.strip().lower()

email = email.strip().lower()

plain_key = encryption_key_plain or generate_user_fernet_key()

encrypted_key = encrypt_for_storage(plain_key)

with db_session() as conn:

cur = conn.execute(

"""

INSERT INTO users (username, email, name, password_hash, encryption_key, created_at, is_active)

VALUES (?, ?, ?, ?, ?, ?, 1)

""",

(username, email, name, password_hash, encrypted_key, utc_now_iso()),

)

user_id = int(cur.lastrowid)

user = get_user_by_id(user_id)

if user is None:

raise RuntimeError("Failed to load user after insert.")

return user

def deactivate_user(user_id: int) -> None:

with db_session() as conn:

conn.execute(

"UPDATE users SET is_active = 0 WHERE user_id = ?",

(user_id,),

)

def username_exists(username: str) -> bool:

return get_user_by_username(username) is not None

def email_exists(email: str) -> bool:

return get_user_by_email(email) is not None