# scopes.api.auth user from datetime import datetime from fastapi import APIRouter, Body import jwt from pwdlib import PasswordHash from pydantic import BaseModel, Field from typing import Annotated, List from scopes.api import deps from scopes.config import settings from scopes.storage import user as st_user from scopes.storage.common import Storage class User(BaseModel): uid: str | None = None timeStamp: datetime | None = None name: str login: str display_name: str email: str | None = None pw_hash: str | None = Field(exclude=True, default=None) groups: List[str] = Field(default_factory=list) class UserIn(User): password: str | None = None def hashPassword(self) -> str | None: pw = self.password return pw and hashPassword(pw) or None # password handling, authentication pw_hash = PasswordHash.recommended() def hashPassword(pw): return pw_hash.hash(pw) def verifyPassword(plain, hashed): return pw_hash.verify(plain, hashed) def getUser(storage: Storage, userName: str) -> User: users = storage.create(st_user.Users) user = users.queryLast(name=userName) return user.toObject(User) async def authenticate(storage: deps.Storage, token: deps.Auth): payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm]) userName = payload.get('sub') user = getUser(storage, userName) return user # routes router = APIRouter() @router.get('/users', response_model=list[User]) async def get_users(storage: deps.Storage): users = storage.create(st_user.Users) result = [tr.toObject(User) for tr in users.query()] return result @router.post('/users', response_model=User) async def post_user(user: Annotated[UserIn, Body()], storage: deps.Storage): users = storage.create(st_user.Users) tr = st_user.User.fromObject(user) users.save(tr) return tr.toObject(User)