81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from app.database import get_db
|
|
from app.dependencies import get_current_user
|
|
from app.schemas.auth import LoginRequest, TokenResponse
|
|
from app.schemas.user import UserCreate, UserResponse, UserWithToken
|
|
from app.services.auth_service import authenticate_user, create_user, create_tokens_for_user
|
|
from app.models.user import User
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["authentication"])
|
|
|
|
|
|
@router.post("/register", response_model=UserWithToken, status_code=status.HTTP_201_CREATED)
|
|
async def register(
|
|
user_data: UserCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Register a new user.
|
|
Registration must be enabled via ALLOW_REGISTRATION environment variable.
|
|
"""
|
|
user = create_user(db, user_data)
|
|
tokens = create_tokens_for_user(user)
|
|
|
|
return UserWithToken(
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
is_admin=user.is_admin,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at,
|
|
access_token=tokens.access_token,
|
|
refresh_token=tokens.refresh_token,
|
|
token_type=tokens.token_type
|
|
)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
async def login(
|
|
credentials: LoginRequest,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Login with username and password.
|
|
Returns JWT access and refresh tokens.
|
|
"""
|
|
user = authenticate_user(db, credentials.username, credentials.password)
|
|
tokens = create_tokens_for_user(user)
|
|
|
|
return tokens
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
async def get_current_user_info(
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Get current authenticated user information.
|
|
Requires valid JWT token in Authorization header.
|
|
"""
|
|
return UserResponse(
|
|
id=current_user.id,
|
|
username=current_user.username,
|
|
email=current_user.email,
|
|
is_admin=current_user.is_admin,
|
|
created_at=current_user.created_at,
|
|
updated_at=current_user.updated_at
|
|
)
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
async def refresh_access_token(
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Refresh access token using a valid refresh token.
|
|
Returns new access and refresh tokens.
|
|
"""
|
|
tokens = create_tokens_for_user(current_user)
|
|
return tokens
|