Dependency to get the current authenticated user (even if not active yet).
(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
)
| 174 | |
| 175 | |
| 176 | async def get_authenticated_user( |
| 177 | credentials: HTTPAuthorizationCredentials = Depends(security), |
| 178 | db: AsyncSession = Depends(get_db), |
| 179 | ): |
| 180 | """Dependency to get the current authenticated user (even if not active yet).""" |
| 181 | from app.models.user import User |
| 182 | |
| 183 | payload = decode_access_token(credentials.credentials) |
| 184 | user_id = payload.get("sub") |
| 185 | if not user_id: |
| 186 | raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") |
| 187 | |
| 188 | result = await db.execute( |
| 189 | select(User) |
| 190 | .where(User.id == uuid.UUID(user_id)) |
| 191 | .options(selectinload(User.identity)) |
| 192 | ) |
| 193 | user = result.scalar_one_or_none() |
| 194 | if not user: |
| 195 | raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") |
| 196 | return user |
| 197 | |
| 198 | |
| 199 | async def get_current_admin(current_user=Depends(get_current_user)): |
nothing calls this directly
no test coverage detected