Coverage for app/api/routes/users.py: 100%
102 statements
« prev ^ index » next coverage.py v7.6.1, created at 2026-03-24 11:40 +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()
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)
43 count = session.exec(count_statement).one()
45 statement = select(User).offset(skip).limit(limit)
46 users = session.exec(statement).all()
48 return UsersPublic(data=users, count=count)
51@router.post(
52 "/", dependencies=[Depends(get_current_active_superuser)], response_model=UserPublic
53)
54def create_user(*, session: SessionDep, user_in: UserCreate) -> Any:
55 """
56 Create new user.
57 """
58 user = crud.get_user_by_email(session=session, email=user_in.email)
59 if user:
60 raise HTTPException(
61 status_code=400,
62 detail="The user with this email already exists in the system.",
63 )
65 user = crud.create_user(session=session, user_create=user_in)
66 if settings.emails_enabled and user_in.email:
67 email_data = generate_new_account_email(
68 email_to=user_in.email, username=user_in.email, password=user_in.password
69 )
70 send_email(
71 email_to=user_in.email,
72 subject=email_data.subject,
73 html_content=email_data.html_content,
74 )
75 return user
78@router.patch("/me", response_model=UserPublic)
79def update_user_me(
80 *, session: SessionDep, user_in: UserUpdateMe, current_user: CurrentUser
81) -> Any:
82 """
83 Update own user.
84 """
86 if user_in.email:
87 existing_user = crud.get_user_by_email(session=session, email=user_in.email)
88 if existing_user and existing_user.id != current_user.id:
89 raise HTTPException(
90 status_code=409, detail="User with this email already exists"
91 )
92 user_data = user_in.model_dump(exclude_unset=True)
93 current_user.sqlmodel_update(user_data)
94 session.add(current_user)
95 session.commit()
96 session.refresh(current_user)
97 return current_user
100@router.patch("/me/password", response_model=Message)
101def update_password_me(
102 *, session: SessionDep, body: UpdatePassword, current_user: CurrentUser
103) -> Any:
104 """
105 Update own password.
106 """
107 if not verify_password(body.current_password, current_user.hashed_password):
108 raise HTTPException(status_code=400, detail="Incorrect password")
109 if body.current_password == body.new_password:
110 raise HTTPException(
111 status_code=400, detail="New password cannot be the same as the current one"
112 )
113 hashed_password = get_password_hash(body.new_password)
114 current_user.hashed_password = hashed_password
115 session.add(current_user)
116 session.commit()
117 return Message(message="Password updated successfully")
120@router.get("/me", response_model=UserPublic)
121def read_user_me(current_user: CurrentUser) -> Any:
122 """
123 Get current user.
124 """
125 return current_user
128@router.delete("/me", response_model=Message)
129def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any:
130 """
131 Delete own user.
132 """
133 if current_user.is_superuser:
134 raise HTTPException(
135 status_code=403, detail="Super users are not allowed to delete themselves"
136 )
137 statement = delete(Item).where(col(Item.owner_id) == current_user.id)
138 session.exec(statement) # type: ignore
139 session.delete(current_user)
140 session.commit()
141 return Message(message="User deleted successfully")
144@router.post("/signup", response_model=UserPublic)
145def register_user(session: SessionDep, user_in: UserRegister) -> Any:
146 """
147 Create new user without the need to be logged in.
148 """
149 user = crud.get_user_by_email(session=session, email=user_in.email)
150 if user:
151 raise HTTPException(
152 status_code=400,
153 detail="The user with this email already exists in the system",
154 )
155 user_create = UserCreate.model_validate(user_in)
156 user = crud.create_user(session=session, user_create=user_create)
157 return user
160@router.get("/{user_id}", response_model=UserPublic)
161def read_user_by_id(
162 user_id: uuid.UUID, session: SessionDep, current_user: CurrentUser
163) -> Any:
164 """
165 Get a specific user by id.
166 """
167 user = session.get(User, user_id)
168 if user == current_user:
169 return user
170 if not current_user.is_superuser:
171 raise HTTPException(
172 status_code=403,
173 detail="The user doesn't have enough privileges",
174 )
175 return user
178@router.patch(
179 "/{user_id}",
180 dependencies=[Depends(get_current_active_superuser)],
181 response_model=UserPublic,
182)
183def update_user(
184 *,
185 session: SessionDep,
186 user_id: uuid.UUID,
187 user_in: UserUpdate,
188) -> Any:
189 """
190 Update a user.
191 """
193 db_user = session.get(User, user_id)
194 if not db_user:
195 raise HTTPException(
196 status_code=404,
197 detail="The user with this id does not exist in the system",
198 )
199 if user_in.email:
200 existing_user = crud.get_user_by_email(session=session, email=user_in.email)
201 if existing_user and existing_user.id != user_id:
202 raise HTTPException(
203 status_code=409, detail="User with this email already exists"
204 )
206 db_user = crud.update_user(session=session, db_user=db_user, user_in=user_in)
207 return db_user
210@router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)])
211def delete_user(
212 session: SessionDep, current_user: CurrentUser, user_id: uuid.UUID
213) -> Message:
214 """
215 Delete a user.
216 """
217 user = session.get(User, user_id)
218 if not user:
219 raise HTTPException(status_code=404, detail="User not found")
220 if user == current_user:
221 raise HTTPException(
222 status_code=403, detail="Super users are not allowed to delete themselves"
223 )
224 statement = delete(Item).where(col(Item.owner_id) == user_id)
225 session.exec(statement) # type: ignore
226 session.delete(user)
227 session.commit()
228 return Message(message="User deleted successfully")