Coverage for fss\starter\system\api\v1\user_controller.py: 91%
45 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-13 15:26 +0800
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-13 15:26 +0800
1"""User operation controller"""
3from fastapi import APIRouter, Depends, UploadFile
4from fastapi.security import OAuth2PasswordRequestForm
5from fastapi_pagination import Params
6from starlette.responses import StreamingResponse
8from fss.common.result import result
9from fss.common.result.result import BaseResponse
10from fss.common.schema.schema import Token, CurrentUser
11from fss.common.security.security import get_current_user
12from fss.common.util.security import get_password_hash
13from fss.starter.system.model.user_do import UserDO
14from fss.starter.system.schema.user_schema import (
15 UserCreateCmd,
16 UserQuery,
17 LoginCmd,
18 UpdateUserCmd,
19)
20from fss.starter.system.service.impl.user_service_impl import get_user_service
21from fss.starter.system.service.user_service import UserService
23user_router = APIRouter()
26@user_router.post("/register")
27async def register_user(
28 create_data: UserCreateCmd, user_service: UserService = Depends(get_user_service)
29) -> BaseResponse[int]:
30 """
31 User registration
32 """
33 create_data.password = await get_password_hash(create_data.password)
34 user: UserDO = await user_service.save(data=create_data)
35 return result.success(data=user.id)
38@user_router.get("/me")
39async def get_user(
40 user_service: UserService = Depends(get_user_service),
41 current_user: CurrentUser = Depends(get_current_user()),
42) -> BaseResponse[UserQuery]:
43 """
44 Query user info
45 """
46 user: UserQuery = await user_service.find_by_id(id=current_user.user_id)
47 return result.success(data=user)
50@user_router.post("/login")
51async def login(
52 login_form: OAuth2PasswordRequestForm = Depends(),
53 user_service: UserService = Depends(get_user_service),
54) -> Token:
55 """
56 User login
57 """
58 loginCmd = LoginCmd(username=login_form.username, password=login_form.password)
59 return await user_service.login(loginCmd)
62@user_router.delete("/{id}")
63async def remove_user(
64 id: int,
65 user_service: UserService = Depends(get_user_service),
66 current_user: CurrentUser = Depends(get_current_user()),
67) -> None:
68 """
69 Remove user
70 """
71 await user_service.remove_by_id(id=id)
72 return result.success()
75@user_router.put("/")
76async def update_user(
77 updateUserCmd: UpdateUserCmd,
78 user_service: UserService = Depends(get_user_service),
79 current_user: CurrentUser = Depends(get_current_user()),
80) -> None:
81 """
82 Update user
83 """
84 await user_service.update_by_id(data=updateUserCmd)
85 return result.success()
88@user_router.get("/exportTemplate")
89async def export_user_template(
90 user_service: UserService = Depends(get_user_service),
91 current_user: CurrentUser = Depends(get_current_user()),
92) -> StreamingResponse:
93 """
94 Export user template
95 """
96 return await user_service.export_user_template()
99@user_router.post("/import")
100async def import_user(
101 file: UploadFile,
102 user_service: UserService = Depends(get_user_service),
103 current_user: CurrentUser = Depends(get_current_user()),
104) -> None:
105 """
106 Import user info
107 """
108 await user_service.import_user(file)
109 return result.success()
112@user_router.get("/export")
113async def export_user(
114 params: Params = Depends(),
115 user_service: UserService = Depends(get_user_service),
116 current_user: CurrentUser = Depends(get_current_user()),
117) -> StreamingResponse:
118 """
119 Export user info
120 """
121 return await user_service.export_user(params)