Skip to content

Users

create_user(*, user_in, service, current_admin) async

Create new user. Admin only.

Parameters:

  • user_in (UserCreate) –

    The user creation data.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_admin (Annotated[User, Depends(get_current_admin)]) –

    The currently authenticated admin user.

Returns:

  • Any ( Any ) –

    The created user object.

Source code in app/api/v1/endpoints/users.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@router.post("/", response_model=UserRead)
async def create_user(
    *,
    user_in: UserCreate,
    service: Annotated[UserService, Depends(get_user_service)],
    current_admin: Annotated[User, Depends(get_current_admin)],
) -> Any:
    """
    Create new user. Admin only.

    Args:
        user_in: The user creation data.
        service: The user service instance.
        current_admin: The currently authenticated admin user.

    Returns:
        Any: The created user object.
    """
    return await service.create(user_in)

delete_user(*, user_id, service, current_admin) async

Delete a user. Admin only.

Parameters:

  • user_id (UUID) –

    The ID of the user to delete.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_admin (Annotated[User, Depends(get_current_admin)]) –

    The currently authenticated admin user.

Returns:

  • Any ( Any ) –

    The deleted user object.

Source code in app/api/v1/endpoints/users.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@router.delete("/{user_id}", response_model=UserRead)
async def delete_user(
    *,
    user_id: UUID,
    service: Annotated[UserService, Depends(get_user_service)],
    current_admin: Annotated[User, Depends(get_current_admin)],
) -> Any:
    """
    Delete a user. Admin only.

    Args:
        user_id: The ID of the user to delete.
        service: The user service instance.
        current_admin: The currently authenticated admin user.

    Returns:
        Any: The deleted user object.
    """
    return await service.delete(user_id)

delete_user_avatar(*, user_id, service, current_user) async

Remove the avatar image from a user.

Parameters:

  • user_id (UUID) –

    The ID of the user.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_user (Annotated[User, Depends(get_current_user)]) –

    The currently authenticated user.

Raises:

  • HTTPException –

    If the user is not authorized.

Returns:

  • Any ( Any ) –

    The updated user object.

Source code in app/api/v1/endpoints/users.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@router.delete("/{user_id}/avatar", response_model=UserRead)
async def delete_user_avatar(
    *,
    user_id: UUID,
    service: Annotated[UserService, Depends(get_user_service)],
    current_user: Annotated[User, Depends(get_current_user)],
) -> Any:
    """
    Remove the avatar image from a user.

    Args:
        user_id: The ID of the user.
        service: The user service instance.
        current_user: The currently authenticated user.

    Raises:
        HTTPException: If the user is not authorized.

    Returns:
        Any: The updated user object.
    """
    if current_user.id != user_id and current_user.role != UserRole.ADMIN:
        raise HTTPException(status_code=403, detail="Not authorized")

    return await service.remove_avatar(user_id)

get_user_avatar(*, user_id, service) async

Get the avatar image file for a user.

Parameters:

  • user_id (UUID) –

    The ID of the user.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

Raises:

  • HTTPException –

    If the avatar is not found.

Returns:

  • FileResponse ( Any ) –

    The avatar image file.

Source code in app/api/v1/endpoints/users.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@router.get("/{user_id}/avatar")
async def get_user_avatar(
    *,
    user_id: UUID,
    service: Annotated[UserService, Depends(get_user_service)],
) -> Any:
    """
    Get the avatar image file for a user.

    Args:
        user_id: The ID of the user.
        service: The user service instance.

    Raises:
        HTTPException: If the avatar is not found.

    Returns:
        FileResponse: The avatar image file.
    """
    file_path = await service.get_avatar_path(user_id)
    if not file_path:
        raise HTTPException(status_code=404, detail="Avatar not found")

    return FileResponse(file_path)

read_user_me(current_user) async

Get current user.

Parameters:

  • current_user (Annotated[User, Depends(get_current_user)]) –

    The currently authenticated user.

Returns:

  • Any ( Any ) –

    The current user object.

Source code in app/api/v1/endpoints/users.py 
18
19
20
21
22
23
24
25
26
27
28
29
@router.get("/me", response_model=UserRead)
async def read_user_me(current_user: Annotated[User, Depends(get_current_user)]) -> Any:
    """
    Get current user.

    Args:
        current_user: The currently authenticated user.

    Returns:
        Any: The current user object.
    """
    return current_user

read_users(service, current_admin, skip=0, limit=100) async

Retrieve users. Admin only.

Parameters:

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_admin (Annotated[User, Depends(get_current_admin)]) –

    The currently authenticated admin user.

  • skip (int, default: 0 ) –

    Number of records to skip.

  • limit (int, default: 100 ) –

    Maximum number of records to return.

Returns:

  • Any ( Any ) –

    A list of users.

Source code in app/api/v1/endpoints/users.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@router.get("/", response_model=List[UserRead])
async def read_users(
    service: Annotated[UserService, Depends(get_user_service)],
    current_admin: Annotated[User, Depends(get_current_admin)],
    skip: int = 0,
    limit: int = 100,
) -> Any:
    """
    Retrieve users. Admin only.

    Args:
        service: The user service instance.
        current_admin: The currently authenticated admin user.
        skip: Number of records to skip.
        limit: Maximum number of records to return.

    Returns:
        Any: A list of users.
    """
    return await service.get_multi(skip=skip, limit=limit)

update_user(*, user_id, user_in, service, current_admin) async

Update a user. Admin only.

Parameters:

  • user_id (UUID) –

    The ID of the user to update.

  • user_in (UserUpdate) –

    The user update data.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_admin (Annotated[User, Depends(get_current_admin)]) –

    The currently authenticated admin user.

Returns:

  • Any ( Any ) –

    The updated user object.

Source code in app/api/v1/endpoints/users.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@router.patch("/{user_id}", response_model=UserRead)
async def update_user(
    *,
    user_id: UUID,
    user_in: UserUpdate,
    service: Annotated[UserService, Depends(get_user_service)],
    current_admin: Annotated[User, Depends(get_current_admin)],
) -> Any:
    """
    Update a user. Admin only.

    Args:
        user_id: The ID of the user to update.
        user_in: The user update data.
        service: The user service instance.
        current_admin: The currently authenticated admin user.

    Returns:
        Any: The updated user object.
    """
    return await service.update(user_id, user_in)

upload_user_avatar(*, user_id, file=File(...), service, current_user) async

Upload an avatar image for a user.

Parameters:

  • user_id (UUID) –

    The ID of the user.

  • file (UploadFile, default: File(...) ) –

    The avatar image file to upload.

  • service (Annotated[UserService, Depends(get_user_service)]) –

    The user service instance.

  • current_user (Annotated[User, Depends(get_current_user)]) –

    The currently authenticated user.

Raises:

  • HTTPException –

    If the user is not authorized.

Returns:

  • Any ( Any ) –

    The updated user object.

Source code in app/api/v1/endpoints/users.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@router.post("/{user_id}/avatar", response_model=UserRead)
async def upload_user_avatar(
    *,
    user_id: UUID,
    file: UploadFile = File(...),
    service: Annotated[UserService, Depends(get_user_service)],
    current_user: Annotated[User, Depends(get_current_user)],
) -> Any:
    """
    Upload an avatar image for a user.

    Args:
        user_id: The ID of the user.
        file: The avatar image file to upload.
        service: The user service instance.
        current_user: The currently authenticated user.

    Raises:
        HTTPException: If the user is not authorized.

    Returns:
        Any: The updated user object.
    """
    if current_user.id != user_id and current_user.role != UserRole.ADMIN:
        raise HTTPException(status_code=403, detail="Not authorized")

    return await service.upload_avatar(user_id, file)