"""Authentication endpoints.""" from datetime import datetime, timedelta from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app import crud, schemas from app.dependencies import get_db, get_current_user from app.core.security import create_access_token from app.config import settings from app.models.user import User router = APIRouter() @router.post("/register", response_model=schemas.User, status_code=status.HTTP_201_CREATED) def register( *, db: Session = Depends(get_db), user_in: schemas.RegisterRequest ) -> Any: """ Register a new user. Creates a new user account with the provided credentials. Registration can be disabled by administrators via settings. """ # Check if this is the first user (always allow for initial setup) user_count = db.query(User).count() is_first_user = user_count == 0 # If not the first user, check if registration is enabled if not is_first_user: registration_enabled = crud.settings.get_setting_value( db, key="registration_enabled", default=True # Default to enabled if setting doesn't exist ) if not registration_enabled: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="User registration is currently disabled" ) # Check if username already exists user = crud.user.get_by_username(db, username=user_in.username) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered" ) # Check if email already exists user = crud.user.get_by_email(db, email=user_in.email) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered" ) # Create new user (first user becomes superuser) user_create = schemas.UserCreate( username=user_in.username, email=user_in.email, password=user_in.password, is_active=True, is_superuser=is_first_user # First user is superuser ) user = crud.user.create(db, obj_in=user_create) return user @router.post("/login", response_model=schemas.Token) def login( *, db: Session = Depends(get_db), credentials: schemas.LoginRequest ) -> Any: """ Login and get access token. Authenticates user and returns a JWT access token. """ # Authenticate user user = crud.user.authenticate( db, username=credentials.username, password=credentials.password ) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) if not crud.user.is_active(user): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user" ) # Update last_login timestamp user.last_login = datetime.utcnow() db.add(user) db.commit() # Create access token access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.id}, expires_delta=access_token_expires ) return { "access_token": access_token, "token_type": "bearer" } @router.get("/me", response_model=schemas.User) def read_users_me( current_user: User = Depends(get_current_user) ) -> Any: """ Get current user. Returns the currently authenticated user's information. Requires a valid JWT token in the Authorization header. """ return current_user @router.get("/registration-status") def get_registration_status( db: Session = Depends(get_db) ) -> Any: """ Check if user registration is enabled. This is a public endpoint that doesn't require authentication. Returns the registration_enabled setting value. """ registration_enabled = crud.settings.get_setting_value( db, key="registration_enabled", default=True ) return {"registration_enabled": registration_enabled}