58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, Optional
|
|
|
|
from .. import config
|
|
from ..schemas.auth import UserInDB, User
|
|
from .security import get_password_hash, verify_password
|
|
|
|
|
|
class UserService:
|
|
"""Simple file-backed user management."""
|
|
|
|
def __init__(self, file_path: Path = config.USERS_FILE) -> None:
|
|
self._file_path = file_path
|
|
self._ensure_default_user()
|
|
|
|
def _ensure_default_user(self) -> None:
|
|
if self._file_path.exists():
|
|
return
|
|
default_user = {
|
|
"users": [
|
|
{
|
|
"username": "admin",
|
|
"hashed_password": get_password_hash("admin123"),
|
|
}
|
|
]
|
|
}
|
|
self._file_path.write_text(json.dumps(default_user, indent=2), encoding="utf-8")
|
|
|
|
def _load_users(self) -> Dict[str, str]:
|
|
data = json.loads(self._file_path.read_text(encoding="utf-8"))
|
|
return {item["username"]: item["hashed_password"] for item in data.get("users", [])}
|
|
|
|
def _save_users(self, users: Dict[str, str]) -> None:
|
|
payload = {"users": [{"username": u, "hashed_password": pw} for u, pw in users.items()]}
|
|
self._file_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
|
|
def get_user(self, username: str) -> Optional[UserInDB]:
|
|
users = self._load_users()
|
|
hashed = users.get(username)
|
|
if not hashed:
|
|
return None
|
|
return UserInDB(username=username, hashed_password=hashed)
|
|
|
|
def authenticate(self, username: str, password: str) -> Optional[User]:
|
|
user = self.get_user(username)
|
|
if not user or not verify_password(password, user.hashed_password):
|
|
return None
|
|
return User(username=user.username)
|
|
|
|
def set_password(self, username: str, password: str) -> User:
|
|
users = self._load_users()
|
|
users[username] = get_password_hash(password)
|
|
self._save_users(users)
|
|
return User(username=username)
|