Spaces:
Sleeping
Sleeping
File size: 4,657 Bytes
d5ea479 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
import uuid
from datetime import datetime
from database.session import get_db
from database.models import User
from schemas import UserCreate, User as UserSchema, Token, UserLogin
from auth_utils import get_password_hash, verify_password, create_access_token, SECRET_KEY, ALGORITHM
router = APIRouter(prefix="/auth", tags=["Authentication"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
result = await db.execute(select(User).where(User.id == uuid.UUID(user_id)))
user = result.scalars().first()
if user is None:
raise credentials_exception
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return user
@router.post("/register", response_model=UserSchema, status_code=status.HTTP_201_CREATED)
async def register(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
# Check if user exists
result = await db.execute(select(User).where(User.email == user_in.email))
if result.scalars().first():
raise HTTPException(
status_code=400,
detail="User with this email already exists"
)
result = await db.execute(select(User).where(User.username == user_in.username))
if result.scalars().first():
raise HTTPException(
status_code=400,
detail="User with this username already exists"
)
# Create user
db_user = User(
email=user_in.email,
username=user_in.username,
name=user_in.name,
password_hash=get_password_hash(user_in.password),
role="USER",
status="active",
balance=0.0000,
spent=0.0000
)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
@router.post("/login", response_model=Token)
async def login(user_in: UserLogin, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.email == user_in.email))
user = result.scalars().first()
if not user or not verify_password(user_in.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
access_token = create_access_token(subject=user.id)
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
# Supports standard OAuth2 form login (username/password)
result = await db.execute(select(User).where(User.username == form_data.username))
user = result.scalars().first()
if not user or not verify_password(form_data.password, user.password_hash):
# Fallback to checking by email if username check failed
result = await db.execute(select(User).where(User.email == form_data.username))
user = result.scalars().first()
if not user or not verify_password(form_data.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(subject=user.id)
return {"access_token": access_token, "token_type": "bearer"}
@router.get("/me", response_model=UserSchema)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
|