87 lines
2.2 KiB
Python
87 lines
2.2 KiB
Python
|
"""
|
||
|
|
Auth API Router - Login, Logout, Me endpoints
|
||
|
|
"""
|
||
|
|
from fastapi import APIRouter, HTTPException, status, Request, Depends
|
||
|
|
from pydantic import BaseModel
|
||
|
|
from app.core.auth_service import AuthService
|
||
|
|
from app.core.auth_dependencies import get_current_user
|
||
|
|
import logging
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
router = APIRouter()
|
||
|
|
|
||
|
|
|
||
|
|
class LoginRequest(BaseModel):
|
||
|
|
username: str
|
||
|
|
password: str
|
||
|
|
|
||
|
|
|
||
|
|
class LoginResponse(BaseModel):
|
||
|
|
access_token: str
|
||
|
|
token_type: str = "bearer"
|
||
|
|
user: dict
|
||
|
|
|
||
|
|
|
||
|
|
class LogoutRequest(BaseModel):
|
||
|
|
token_jti: str
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/login", response_model=LoginResponse)
|
||
|
|
async def login(request: Request, credentials: LoginRequest):
|
||
|
|
"""
|
||
|
|
Authenticate user and return JWT token
|
||
|
|
"""
|
||
|
|
ip_address = request.client.host if request.client else None
|
||
|
|
|
||
|
|
# Authenticate user
|
||
|
|
user = AuthService.authenticate_user(
|
||
|
|
username=credentials.username,
|
||
|
|
password=credentials.password,
|
||
|
|
ip_address=ip_address
|
||
|
|
)
|
||
|
|
|
||
|
|
if not user:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Invalid username or password",
|
||
|
|
headers={"WWW-Authenticate": "Bearer"},
|
||
|
|
)
|
||
|
|
|
||
|
|
# Create access token
|
||
|
|
access_token = AuthService.create_access_token(
|
||
|
|
user_id=user['user_id'],
|
||
|
|
username=user['username'],
|
||
|
|
is_superadmin=user['is_superadmin']
|
||
|
|
)
|
||
|
|
|
||
|
|
return LoginResponse(
|
||
|
|
access_token=access_token,
|
||
|
|
user=user
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/logout")
|
||
|
|
async def logout(request: LogoutRequest, current_user: dict = Depends(get_current_user)):
|
||
|
|
"""
|
||
|
|
Revoke JWT token (logout)
|
||
|
|
"""
|
||
|
|
AuthService.revoke_token(request.token_jti, current_user['id'])
|
||
|
|
|
||
|
|
return {"message": "Successfully logged out"}
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/me")
|
||
|
|
async def get_me(current_user: dict = Depends(get_current_user)):
|
||
|
|
"""
|
||
|
|
Get current authenticated user info
|
||
|
|
"""
|
||
|
|
return {
|
||
|
|
"id": current_user['id'],
|
||
|
|
"username": current_user['username'],
|
||
|
|
"email": current_user['email'],
|
||
|
|
"full_name": current_user['full_name'],
|
||
|
|
"is_superadmin": current_user['is_superadmin'],
|
||
|
|
"permissions": current_user['permissions']
|
||
|
|
}
|