Leopoldadmin
d89304845a
- Backend: FastAPI, JWT auth (HS256), bcrypt, RBAC middleware - User CRUD (admin-only), soft-delete, pagination - Pydantic BaseSettings config, async SQLAlchemy 2.0 - 50 backend tests, 88% coverage - Frontend: Next.js 14 App Router, Tailwind, design tokens - Login page, auth context, API client with auto-refresh - i18n: next-intl, DE/EN (31 keys each) - 6 base UI components (Button, Input, Card, Table, Modal, Toast) - 12 frontend tests, npm build success
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""Auth router: login, token refresh, current user profile."""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.dependencies import get_current_user
|
|
from app.models.user import User
|
|
from app.schemas.user import (
|
|
LoginRequest,
|
|
RefreshRequest,
|
|
TokenResponse,
|
|
UserResponse,
|
|
)
|
|
from app.services.auth_service import (
|
|
authenticate_user,
|
|
generate_token_pair,
|
|
refresh_access_token,
|
|
)
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse, status_code=status.HTTP_200_OK)
|
|
async def login(
|
|
body: LoginRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Authenticate user and return JWT access + refresh tokens."""
|
|
user = await authenticate_user(db, body.email, body.password)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "INVALID_CREDENTIALS",
|
|
"message": "Invalid email or password",
|
|
}
|
|
},
|
|
)
|
|
tokens = generate_token_pair(user)
|
|
return TokenResponse(**tokens)
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse, status_code=status.HTTP_200_OK)
|
|
async def refresh_token(
|
|
body: RefreshRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Exchange a valid refresh token for a new access + refresh token pair."""
|
|
tokens = await refresh_access_token(db, body.refresh_token)
|
|
if tokens is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "INVALID_REFRESH_TOKEN",
|
|
"message": "Invalid or expired refresh token",
|
|
}
|
|
},
|
|
)
|
|
return TokenResponse(**tokens)
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse, status_code=status.HTTP_200_OK)
|
|
async def get_me(current_user: User = Depends(get_current_user)):
|
|
"""Return the currently authenticated user's profile."""
|
|
return UserResponse.model_validate(current_user)
|