Download src/api/v1/users.py from DataEyond/Agentic-Service-Data-Eyond-Catalog: direct link, hf CLI and curl.
- Browser
- Download file 2.1 kB
-
https://huggingface.co/spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog/resolve/717452f69ec32394dea2a8fc11aba47eebe5bc53/src/api/v1/users.py
- Command line
-
hf download hf://spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog@717452f69ec32394dea2a8fc11aba47eebe5bc53/src/api/v1/users.py
-
curl -L -o users.py https://huggingface.co/spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog/resolve/717452f69ec32394dea2a8fc11aba47eebe5bc53/src/api/v1/users.py
2.1 kB
| import pandas as pd | |
| from datetime import datetime | |
| from fastapi.responses import JSONResponse | |
| from fastapi import APIRouter, HTTPException, status | |
| from typing import Literal | |
| from src.users.users import get_user, hash_password, verify_password | |
| from src.middlewares.logging import get_logger | |
| from pydantic import BaseModel | |
| class ILogin(BaseModel): | |
| """Login request model.""" | |
| email: str | |
| password: str | |
| logger = get_logger("users service") | |
| router = APIRouter( | |
| prefix="/api", | |
| tags=["Users"], | |
| ) | |
| from typing import Optional, Literal | |
| async def login(payload: ILogin): | |
| """ | |
| Authenticates a user and returns their data if credentials are valid. | |
| """ | |
| try: | |
| user_profile:dict | None= await get_user(payload.email) | |
| except Exception as E: | |
| print(f"❌ login error while fetching user: {E}") | |
| # Return generic 500 to client | |
| raise HTTPException( | |
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, | |
| detail="Internal server error" | |
| ) | |
| if not user_profile: | |
| # 404 or 401 – choose based on your security policy | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail="Email not found" | |
| ) | |
| if user_profile.get("status") == "inactive": | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="Account is inactive" | |
| ) | |
| is_verified = verify_password( | |
| password=payload.password, | |
| hashed_password=user_profile.get("password") | |
| ) | |
| if not is_verified: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Email or password invalid" | |
| ) | |
| user_profile.pop("password", None) | |
| return { | |
| "status": "success", | |
| "message": "success", | |
| "data": user_profile, | |
| } | |