Spaces:
Running
Running
Anish
[Updated Feature] > Updated every file_route, user_route, so that guest users can login and only upload 5 videos, 1 image per IP
e000e33 | from fastapi import Depends, HTTPException | |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
| from jose import jwt, JWTError | |
| from app.core.config import settings | |
| from sqlalchemy.orm import Session | |
| from app.db.session import get_db | |
| from app.models.user_model import User | |
| from typing import Optional | |
| security = HTTPBearer() | |
| optional_security = HTTPBearer(auto_error=False) | |
| def get_current_user( | |
| credentials: HTTPAuthorizationCredentials = Depends(security), | |
| db: Session = Depends(get_db) | |
| ): | |
| token = credentials.credentials | |
| try: | |
| payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"]) | |
| user_id: int = payload.get("user_id") | |
| if user_id is None: | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| except JWTError: | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| user = db.query(User).filter(User.id == user_id).first() | |
| if user is None: | |
| raise HTTPException(status_code=401, detail="User not found") | |
| return user | |
| def get_optional_user( | |
| credentials: | |
| Optional[HTTPAuthorizationCredentials] = Depends(optional_security), | |
| db: Session = Depends(get_db) | |
| ): | |
| if not credentials: | |
| return None | |
| token = credentials.credentials | |
| try: | |
| payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"]) | |
| user_id = payload.get("user_id") | |
| if not user_id: | |
| return None | |
| return db.query(User).filter(User.id == user_id).first() | |
| except JWTError: | |
| return None |