From 6b6f20a032eb31d747366c8bc270ca81731eaf6a Mon Sep 17 00:00:00 2001 From: Tom van der Lee Date: Fri, 4 Sep 2026 17:09:24 +0200 Subject: Fixed pre-commit issues --- authentication/backend.py | 49 ++++++++++++++++++++--------------------------- 1 file changed, 21 insertions(+), 28 deletions(-) (limited to 'authentication/backend.py') diff --git a/authentication/backend.py b/authentication/backend.py index 9434906..58842de 100644 --- a/authentication/backend.py +++ b/authentication/backend.py @@ -1,52 +1,49 @@ import re -from typing import Optional from sqlalchemy import update from sqlmodel import select -from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser +from starlette.authentication import ( + AuthCredentials, + AuthenticationBackend, + BaseUser, + UnauthenticatedUser, +) from starlette.requests import HTTPConnection -from db.models import Session, User, AuthToken +from db.models import AuthToken, Session, User from db.session import get_session_context class SessionAuthBackend(AuthenticationBackend): - async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: + async def authenticate( + self, request: HTTPConnection + ) -> tuple[AuthCredentials, BaseUser] | None: if "id" not in request.session: return None - query = ( - select(Session) - .join(User) - .where( - Session.id == request.session['id'] - ) - ) - + query = select(Session).join(User).where(Session.id == request.session["id"]) with get_session_context() as db_session: session: Session = db_session.exec(query).first() if session is not None: - db_session.execute( - update(Session).where(Session.id == session.id) - ) + db_session.execute(update(Session).where(Session.id == session.id)) db_session.commit() - return ( AuthCredentials( - ["authenticated", "session"] - if session is not None - else [] + ["authenticated", "session"] if session is not None else [] ), - session.user if session is not None else UnauthenticatedUser() + session.user if session is not None else UnauthenticatedUser(), ) class BearerTokenAuthBackend(AuthenticationBackend): regex = re.compile(r"^[Bb]earer\s(?P\S+)$") - async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: + + async def authenticate( + self, request: HTTPConnection + ) -> tuple[AuthCredentials, BaseUser] | None: token = None if "Authorization" in request.headers: match = self.regex.match(request.headers["Authorization"]) @@ -54,17 +51,13 @@ class BearerTokenAuthBackend(AuthenticationBackend): query = ( select(AuthToken) .join(User) - .where(AuthToken.access_token == match.group('token')) + .where(AuthToken.access_token == match.group("token")) ) with get_session_context() as db_session: token: AuthToken = db_session.exec(query).first() return ( - AuthCredentials( - ["authenticated", "token"] - if token is not None - else [] - ), - token.user if token is not None else UnauthenticatedUser() + AuthCredentials(["authenticated", "token"] if token is not None else []), + token.user if token is not None else UnauthenticatedUser(), ) -- cgit v1.3