-
Notifications
You must be signed in to change notification settings - Fork 3
/
users.py
59 lines (47 loc) · 2.11 KB
/
users.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
from fastapi import APIRouter
from fastapi import Depends
from sqlalchemy.orm import Session
from fastapi_pagination import Page
from app.sessions.db import create_local_session
from app.daos.users import (
create_user as create_user_dao,
get_user as get_user_dao,
list_users as list_users_dao,
login as signin,
)
from app.schemas.users.users_request import CreateUser, Login
from app.schemas.users.users_response import UserOutResponse
from app.utils.user_utils import get_current_user
from typing import Annotated
from fastapi.security import HTTPBearer
from app.middlewares.request_id_injection import request_id_contextvar
user_router = APIRouter()
httpBearerScheme = HTTPBearer()
@user_router.post("/register", tags=["Users"])
def register(payload: CreateUser, db: Session = Depends(create_local_session)):
print("Request ID:", request_id_contextvar.get())
response = create_user_dao(data=payload, dbSession=db)
return response
@user_router.post("/signin", tags=["Users"])
def login(payload: Login, db: Session = Depends(create_local_session)):
print("Request ID:", request_id_contextvar.get())
response = signin(data=payload, dbSession=db)
return response
@user_router.get("/{user_id}", tags=["Users"], dependencies=[Depends(get_current_user)], response_model=UserOutResponse)
async def profile(
token: Annotated[str, Depends(httpBearerScheme)],
user_id,
db: Session = Depends(create_local_session),
):
print("Request ID:", request_id_contextvar.get())
response = await get_user_dao(user_id, dbSession=db)
return response
@user_router.get("/", tags=["Users"], response_model=Page[UserOutResponse])
def list_users(db: Session = Depends(create_local_session)):
print("Request ID:", request_id_contextvar.get())
response = list_users_dao(dbSession=db)
return response
@user_router.get("/{user_id}/secure-route/", tags=["Users"], dependencies=[Depends(get_current_user)])
def secure_route(token: Annotated[str, Depends(httpBearerScheme)], user_id: int):
print("Request ID:", request_id_contextvar.get())
return {"message": "If you see this, you're authenticated"}