37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from app.db import Database
|
|
from app.models import utc_now_iso
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, db: Database) -> None:
|
|
self._db = db
|
|
|
|
def get_by_username(self, username: str) -> dict | None:
|
|
with self._db.connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT id, username, password_hash, role FROM users WHERE username = ?",
|
|
(username,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def upsert(self, username: str, password_hash: str, role: str = "admin") -> None:
|
|
now = utc_now_iso()
|
|
with self._db.connect() as conn:
|
|
existing = conn.execute("SELECT id FROM users WHERE username = ?", (username,)).fetchone()
|
|
if existing:
|
|
conn.execute(
|
|
"UPDATE users SET password_hash = ?, role = ?, updated_at = ? WHERE username = ?",
|
|
(password_hash, role, now, username),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO users (username, password_hash, role, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(username, password_hash, role, now, now),
|
|
)
|
|
conn.commit()
|
|
|