2026-04-06 12:41:49 +01:00
|
|
|
"""
|
|
|
|
|
Router dependencies for user module
|
|
|
|
|
|
|
|
|
|
Classes:
|
|
|
|
|
- List: Description
|
|
|
|
|
- Classes: Description
|
|
|
|
|
|
|
|
|
|
Functions:
|
|
|
|
|
- List: Description
|
|
|
|
|
- Functions: Description
|
2026-05-27 13:27:24 +01:00
|
|
|
"""
|
|
|
|
|
from typing import Annotated
|
|
|
|
|
from fastapi import Depends, Query
|
|
|
|
|
|
|
|
|
|
from src.user.exceptions import UserNotFoundException
|
|
|
|
|
from src.user.models import User
|
|
|
|
|
|
|
|
|
|
from src.auth.service import claims_dependency
|
|
|
|
|
from src.database import db_dependency
|
|
|
|
|
from src.user.schemas import UserIDMixin
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_user_model_claims(claims: claims_dependency, db: db_dependency):
|
|
|
|
|
user_id = claims.get("db_id", None)
|
|
|
|
|
if user_id is None:
|
|
|
|
|
raise UserNotFoundException()
|
|
|
|
|
|
|
|
|
|
user_model = db.get(User, user_id)
|
|
|
|
|
if user_model is None:
|
|
|
|
|
raise UserNotFoundException(user_id=user_id)
|
|
|
|
|
|
|
|
|
|
return user_model
|
|
|
|
|
|
|
|
|
|
user_model_claims_dependency = Annotated[type[User], Depends(get_user_model_claims)]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_user_model_query(db: db_dependency, user_id: Annotated[int, Query(gt=0)]):
|
|
|
|
|
user_model = db.get(User, user_id)
|
|
|
|
|
if user_model is None:
|
|
|
|
|
raise UserNotFoundException(user_id=user_id)
|
|
|
|
|
|
|
|
|
|
return user_model
|
|
|
|
|
|
|
|
|
|
user_model_query_dependency = Annotated[type[User], Depends(get_user_model_query)]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_user_model_body(db: db_dependency, request_model: UserIDMixin):
|
|
|
|
|
user_model = db.get(User, request_model.user_id)
|
|
|
|
|
if user_model is None:
|
|
|
|
|
raise UserNotFoundException(user_id=request_model.user_id)
|
|
|
|
|
|
|
|
|
|
return user_model
|
|
|
|
|
|
|
|
|
|
user_model_body_dependency = Annotated[type[User], Depends(get_user_model_body)]
|