Spaces:
Sleeping
Sleeping
File size: 3,926 Bytes
9096e54 | 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 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from datetime import datetime
from app.core import get_db, get_password_hash, verify_password, create_access_token, create_refresh_token
from app.core.security import decode_token
from app.models import User
from app.schemas import LoginRequest, Token, TokenRefresh, UserCreate, UserResponse
router = APIRouter(prefix="/auth", tags=["Authentication"])
@router.post("/login", response_model=Token)
async def login(login_data: LoginRequest, db: Session = Depends(get_db)):
"""User login"""
user = db.query(User).filter(
(User.username == login_data.username) | (User.email == login_data.username)
).first()
if not user or not verify_password(login_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password"
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User account is inactive"
)
# Update last login
user.last_login = datetime.utcnow()
db.commit()
# Create tokens
access_token = create_access_token(data={"sub": str(user.id)})
refresh_token = create_refresh_token(data={"sub": str(user.id)})
return Token(access_token=access_token, refresh_token=refresh_token)
@router.post("/refresh", response_model=Token)
async def refresh_token(token_data: TokenRefresh, db: Session = Depends(get_db)):
"""Refresh access token"""
payload = decode_token(token_data.refresh_token)
if payload.get("type") != "refresh":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token type"
)
user_id = payload.get("sub")
user = db.query(User).filter(User.id == user_id).first()
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or inactive"
)
access_token = create_access_token(data={"sub": str(user.id)})
refresh_token = create_refresh_token(data={"sub": str(user.id)})
return Token(access_token=access_token, refresh_token=refresh_token)
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: Session = Depends(get_db)):
"""Register a new user (for initial setup only)"""
# Check if any users exist
existing_users = db.query(User).count()
# If users exist, only allow superadmin to create new users
if existing_users > 0:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Registration is disabled. Contact admin to create accounts."
)
# Check if username or email already exists
if db.query(User).filter(User.username == user_data.username).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered"
)
if db.query(User).filter(User.email == user_data.email).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Create user (first user becomes superadmin)
user = User(
email=user_data.email,
username=user_data.username,
hashed_password=get_password_hash(user_data.password),
full_name=user_data.full_name,
phone=user_data.phone,
role="superadmin" if existing_users == 0 else user_data.role
)
db.add(user)
db.commit()
db.refresh(user)
return user
|