From 8963c2ac87c720e7372059e7ddb3fa86d79e6e0a Mon Sep 17 00:00:00 2001 From: Helmut Merz Date: Mon, 9 Nov 2026 08:34:33 +0100 Subject: [PATCH] api: improvements, storage users as dependency --- scopes/api/auth/config.py | 2 +- scopes/api/auth/user.py | 17 +++++------------ scopes/api/deps.py | 16 +++++++++++++++- tests/test_api.py | 6 ++---- 4 files changed, 23 insertions(+), 18 deletions(-) diff --git a/scopes/api/auth/config.py b/scopes/api/auth/config.py index 9b63f8d..9dd496e 100644 --- a/scopes/api/auth/config.py +++ b/scopes/api/auth/config.py @@ -9,7 +9,7 @@ class Settings(BaseSettings): #, env_prefix='SCOPES_') auth_method: str = 'oauth2' jwt_key: str | None = None - jwt_atlgorithm: str = 'HS256' + jwt_algorithm: str = 'HS256' settings = Settings() diff --git a/scopes/api/auth/user.py b/scopes/api/auth/user.py index bec98ed..88634c0 100644 --- a/scopes/api/auth/user.py +++ b/scopes/api/auth/user.py @@ -24,7 +24,7 @@ class User(UserBase): timeStamp: datetime | None = None pw_hash: str | None = Field(exclude=True, default=None) -class UserIn(UserBase): +class UserEdit(UserBase): password: str | None = None def hashPassword(self) -> str | None: @@ -41,15 +41,10 @@ def hashPassword(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.AuthUser): +def userFromToken(users: deps.Users, token: deps.AuthToken): payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm]) userName = payload.get('sub') - user = getUser(storage, userName) + user = users.queryLast(name=userName) return user # routes @@ -57,16 +52,14 @@ async def authenticate(storage: deps.Storage, token: deps.AuthUser): router = APIRouter() @router.get('/users', response_model=list[User]) -async def get_users(storage: deps.Storage): - users = storage.create(st_user.Users) +async def get_users(users: deps.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, +async def post_user(user: Annotated[UserEdit, Body()], users: deps.Users, auth_user: deps.AuthUser): print('*** post_user', auth_user) - users = storage.create(st_user.Users) tr = st_user.User.fromObject(user) users.save(tr) return tr.toObject(User) diff --git a/scopes/api/deps.py b/scopes/api/deps.py index 5b2de77..2f9e4a8 100644 --- a/scopes/api/deps.py +++ b/scopes/api/deps.py @@ -8,16 +8,30 @@ from typing import Annotated from scopes import config from scopes.api.auth import config as auth_config from scopes.storage import common +from scopes.storage import user as st_user type Storage = Annotated[common.Storage, Depends(storage)] +type Users = Annotated[st_user.Users, Depends(users)] + type AuthUser = Annotated[str, Depends(auth_user)] +type AuthToken = Annotated[str, Depends(auth_token)] + +# storage def storage(): stf = config.StorageFactory(config.settings) return stf(config.settings.dbschema) +def users(storage: Storage): + return storage.create(st_user.Users) + +# auth + def auth_user(): auth = auth_config.auth_methods.get(auth_config.settings.auth_method) - if auth: + if auth is not None: return auth() +def auth_token(): + return None + diff --git a/tests/test_api.py b/tests/test_api.py index 49bb68a..3818ae0 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,21 +3,19 @@ """Tests for the fastapi- (and asyncio-) based stuff""" from scopes import config +from scopes.api.auth import config as auth_config config.settings = config.Settings(_env_file='tests/test.env') +auth_config.settings = auth_config.Settings(_env_file='tests/test.env') import pytest from fastapi.testclient import TestClient from httpx2 import ASGITransport, AsyncClient, BasicAuth from scopes.api import main -from scopes.api.auth import config as auth_config from scopes.api.main import app from scopes.storage import tracking, user as st_user from scopes.api.auth import user as api_user -config.settings = config.Settings(_env_file='tests/test.env') -auth_config.settings = auth_config.Settings(_env_file='tests/test.env') - @pytest.fixture(scope='module') def storage(): storage = config.StorageFactory(config.settings)(config.settings.dbschema)