forked from sr2/cloud-api
55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
"""
|
|
Dependencies related to the user module
|
|
|
|
Exports:
|
|
- user_model_claims_dependency: user_model: Gets user model from db, if it exists. Uses db_id from user claims.
|
|
- user_model_query_dependency: user_model: Gets user model from db, if it exists. Uses user_id from query param
|
|
- user_model_body_dependency: user_model: Gets user model from db, if it exists. Uses user_id from request body.
|
|
"""
|
|
|
|
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.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[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[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[User, Depends(get_user_model_body)]
|