Spaces:
Sleeping
Sleeping
| from fastapi import HTTPException, status | |
| from src.apis.models.user_models import User | |
| from src.config.mongo import UserCRUD | |
| from src.apis.providers.jwt_provider import JWTProvider | |
| from src.utils.logger import logger | |
| import jwt | |
| jwt_provider = JWTProvider() | |
| async def login_control(username: str, password: str): | |
| user = User(username=username, password=password) | |
| existing_user = await UserCRUD.read_one({"username": user.username}) | |
| if not existing_user: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid username or password", | |
| ) | |
| if existing_user and existing_user["password"] != user.password: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid username or password", | |
| ) | |
| token = jwt_provider.encrypt({"id": str(existing_user["_id"])}) | |
| user_data = user.__dict__ | |
| user_data["id"] = existing_user["_id"] | |
| user_data["username"] = existing_user["username"] | |
| user_data.pop("password", None) | |
| user_data.pop("created_at", None) | |
| user_data.pop("updated_at", None) | |
| user_data.pop("expire_at", None) | |
| return token, user_data | |