55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
"""Authentication service."""
|
|
|
|
from app.interfaces.db_api_client import DBApiClient
|
|
from app.models.auth import LoginRequest, LoginResponse, UserResponse
|
|
from app.utils.security import create_access_token
|
|
|
|
|
|
class AuthService:
|
|
"""Service for user authentication."""
|
|
|
|
def __init__(self, db_client: DBApiClient):
|
|
"""
|
|
Initialize auth service.
|
|
|
|
Args:
|
|
db_client: DB API client instance
|
|
"""
|
|
self.db_client = db_client
|
|
|
|
async def login(self, login: str, client_ip: str) -> LoginResponse:
|
|
"""
|
|
Authenticate user and generate JWT token.
|
|
|
|
Args:
|
|
login: 8-digit login
|
|
client_ip: Client IP address
|
|
|
|
Returns:
|
|
LoginResponse with JWT token and user info
|
|
|
|
Raises:
|
|
ValueError: If login format is invalid
|
|
Exception: If DB API call fails
|
|
"""
|
|
# Validate login format
|
|
if not (login.isdigit() and len(login) == 8):
|
|
raise ValueError("Login must be 8 digits")
|
|
|
|
# Call DB API to validate and record login
|
|
request = LoginRequest(login=login, client_ip=client_ip)
|
|
user: UserResponse = await self.db_client.login_user(request)
|
|
|
|
# Generate JWT token
|
|
token_data = {
|
|
"user_id": user.user_id,
|
|
"login": user.login
|
|
}
|
|
access_token = create_access_token(token_data)
|
|
|
|
return LoginResponse(
|
|
access_token=access_token,
|
|
token_type="bearer",
|
|
user=user
|
|
)
|