Coverage for app/api/routes/users.py: 100%
104 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-26 10:48 +0000
1import uuid
2from typing import Any
4from fastapi import APIRouter, Depends, HTTPException
5from sqlmodel import col, delete, func, select
7from app import crud
8from app.api.deps import (
9 CurrentUser,
10 SessionDep,
11 get_current_active_superuser,
12)
13from app.core.config import settings
14from app.core.security import get_password_hash, verify_password
15from app.models import (
16 Item,
17 Message,
18 UpdatePassword,
19 User,
20 UserCreate,
21 UserPublic,
22 UserRegister,
23 UsersPublic,
24 UserUpdate,
25 UserUpdateMe,
26)
27from app.utils import generate_new_account_email, send_email
29router = APIRouter(prefix="/users", tags=["users"])
32@router.get(
33 "/",
34 dependencies=[Depends(get_current_active_superuser)],
35 response_model=UsersPublic,
36)
37def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
38 """
39 Retrieve users.
40 """
42 count_statement = select(func.count()).select_from(User) 1e
43 count = session.exec(count_statement).one() 1e
45 statement = ( 1e
46 select(User).order_by(col(User.created_at).desc()).offset(skip).limit(limit)
47 )
48 users = session.exec(statement).all() 1e
50 users_public = [UserPublic.model_validate(user) for user in users] 1e
51 return UsersPublic(data=users_public, count=count) 1e
54@router.post(
55 "/", dependencies=[Depends(get_current_active_superuser)], response_model=UserPublic
56)
57def create_user(*, session: SessionDep, user_in: UserCreate) -> Any:
58 """
59 Create new user.
60 """
61 user = crud.get_user_by_email(session=session, email=user_in.email) 1rg
62 if user: 1rg
63 raise HTTPException( 1r
64 status_code=400,
65 detail="The user with this email already exists in the system.",
66 )
68 user = crud.create_user(session=session, user_create=user_in) 1g
69 if settings.emails_enabled and user_in.email: 1g
70 email_data = generate_new_account_email( 1g
71 email_to=user_in.email, username=user_in.email, password=user_in.password
72 )
73 send_email(
74 email_to=user_in.email,
75 subject=email_data.subject,
76 html_content=email_data.html_content,
77 )
78 return user
81@router.patch("/me", response_model=UserPublic)
82def update_user_me(
83 *, session: SessionDep, user_in: UserUpdateMe, current_user: CurrentUser
84) -> Any:
85 """
86 Update own user.
87 """
89 if user_in.email: 1bl
90 existing_user = crud.get_user_by_email(session=session, email=user_in.email) 1bl
91 if existing_user and existing_user.id != current_user.id: 1bl
92 raise HTTPException( 1l
93 status_code=409, detail="User with this email already exists"
94 )
95 user_data = user_in.model_dump(exclude_unset=True) 1b
96 current_user.sqlmodel_update(user_data) 1b
97 session.add(current_user) 1b
98 session.commit() 1b
99 session.refresh(current_user) 1b
100 return current_user 1b
103@router.patch("/me/password", response_model=Message)
104def update_password_me(
105 *, session: SessionDep, body: UpdatePassword, current_user: CurrentUser
106) -> Any:
107 """
108 Update own password.
109 """
110 verified, _ = verify_password(body.current_password, current_user.hashed_password) 1csm
111 if not verified: 1csm
112 raise HTTPException(status_code=400, detail="Incorrect password") 1s
113 if body.current_password == body.new_password: 1cm
114 raise HTTPException( 1m
115 status_code=400, detail="New password cannot be the same as the current one"
116 )
117 hashed_password = get_password_hash(body.new_password) 1c
118 current_user.hashed_password = hashed_password 1c
119 session.add(current_user) 1c
120 session.commit() 1c
121 return Message(message="Password updated successfully") 1c
124@router.get("/me", response_model=UserPublic)
125def read_user_me(current_user: CurrentUser) -> Any:
126 """
127 Get current user.
128 """
129 return current_user 1yz
132@router.delete("/me", response_model=Message)
133def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any:
134 """
135 Delete own user.
136 """
137 if current_user.is_superuser: 1nx
138 raise HTTPException( 1x
139 status_code=403, detail="Super users are not allowed to delete themselves"
140 )
141 session.delete(current_user) 1n
142 session.commit() 1n
143 return Message(message="User deleted successfully") 1n
146@router.post("/signup", response_model=UserPublic)
147def register_user(session: SessionDep, user_in: UserRegister) -> Any:
148 """
149 Create new user without the need to be logged in.
150 """
151 user = crud.get_user_by_email(session=session, email=user_in.email) 1ht
152 if user: 1ht
153 raise HTTPException( 1t
154 status_code=400,
155 detail="The user with this email already exists in the system",
156 )
157 user_create = UserCreate.model_validate(user_in) 1h
158 user = crud.create_user(session=session, user_create=user_create) 1h
159 return user 1h
162@router.get("/{user_id}", response_model=UserPublic)
163def read_user_by_id(
164 user_id: uuid.UUID, session: SessionDep, current_user: CurrentUser
165) -> Any:
166 """
167 Get a specific user by id.
168 """
169 user = session.get(User, user_id) 1iuojp
170 if user == current_user: 1iuojp
171 return user 1u
172 if not current_user.is_superuser: 1iojp
173 raise HTTPException( 1op
174 status_code=403,
175 detail="The user doesn't have enough privileges",
176 )
177 if user is None: 1ij
178 raise HTTPException(status_code=404, detail="User not found") 1j
179 return user 1i
182@router.patch(
183 "/{user_id}",
184 dependencies=[Depends(get_current_active_superuser)],
185 response_model=UserPublic,
186)
187def update_user(
188 *,
189 session: SessionDep,
190 user_id: uuid.UUID,
191 user_in: UserUpdate,
192) -> Any:
193 """
194 Update a user.
195 """
197 db_user = session.get(User, user_id) 1kfv
198 if not db_user: 1kfv
199 raise HTTPException( 1v
200 status_code=404,
201 detail="The user with this id does not exist in the system",
202 )
203 if user_in.email: 1kf
204 existing_user = crud.get_user_by_email(session=session, email=user_in.email) 1f
205 if existing_user and existing_user.id != user_id: 1f
206 raise HTTPException( 1f
207 status_code=409, detail="User with this email already exists"
208 )
210 db_user = crud.update_user(session=session, db_user=db_user, user_in=user_in) 1k
211 return db_user 1k
214@router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)])
215def delete_user(
216 session: SessionDep, current_user: CurrentUser, user_id: uuid.UUID
217) -> Message:
218 """
219 Delete a user.
220 """
221 user = session.get(User, user_id) 1qwd
222 if not user: 1qwd
223 raise HTTPException(status_code=404, detail="User not found") 1w
224 if user == current_user: 1qd
225 raise HTTPException( 1q
226 status_code=403, detail="Super users are not allowed to delete themselves"
227 )
228 statement = delete(Item).where(col(Item.owner_id) == user_id) 1d
229 session.exec(statement) 1d
230 session.delete(user) 1d
231 session.commit() 1d
232 return Message(message="User deleted successfully") 1d