Dependency to get the current authenticated and active user.
(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
)
| 151 | |
| 152 | |
| 153 | async def get_current_user( |
| 154 | credentials: HTTPAuthorizationCredentials = Depends(security), |
| 155 | db: AsyncSession = Depends(get_db), |
| 156 | ): |
| 157 | """Dependency to get the current authenticated and active user.""" |
| 158 | from app.models.user import User |
| 159 | |
| 160 | payload = decode_access_token(credentials.credentials) |
| 161 | user_id = payload.get("sub") |
| 162 | if not user_id: |
| 163 | raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") |
| 164 | |
| 165 | result = await db.execute( |
| 166 | select(User) |
| 167 | .where(User.id == uuid.UUID(user_id)) |
| 168 | .options(selectinload(User.identity)) |
| 169 | ) |
| 170 | user = result.scalar_one_or_none() |
| 171 | if not user or not user.is_active: |
| 172 | raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive") |
| 173 | return user |
| 174 | |
| 175 | |
| 176 | async def get_authenticated_user( |
nothing calls this directly
no test coverage detected