87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""Authentication service: login, register, token management."""
|
|
from sqlalchemy.orm import Session
|
|
from app.models.user import User
|
|
from app.models.tenant import Tenant
|
|
from app.core.security import hash_password, verify_password, create_access_token
|
|
from app.core.exceptions import AppException, NotFoundError, PermissionDeniedError
|
|
from app.schemas.auth import LoginRequest, RegisterRequest
|
|
from app.schemas.common import TokenResponse, UserBrief
|
|
|
|
|
|
class AuthService:
|
|
@staticmethod
|
|
def login(db: Session, req: LoginRequest) -> TokenResponse:
|
|
"""Authenticate user and return JWT token."""
|
|
user = db.query(User).filter(User.email == req.email).first()
|
|
if not user or not user.hashed_password:
|
|
raise AppException("Invalid email or password", 401)
|
|
if not user.is_active:
|
|
raise AppException("Account is disabled", 403)
|
|
if not verify_password(req.password, user.hashed_password):
|
|
raise AppException("Invalid email or password", 401)
|
|
|
|
token = create_access_token(
|
|
subject=str(user.id),
|
|
tenant_id=user.tenant_id,
|
|
role=user.role,
|
|
)
|
|
return TokenResponse(
|
|
access_token=token,
|
|
user=UserBrief.model_validate(user),
|
|
)
|
|
|
|
@staticmethod
|
|
def register(db: Session, req: RegisterRequest) -> TokenResponse:
|
|
"""Register a new user. Creates tenant if tenant_name is provided."""
|
|
# Check if email already exists
|
|
existing = db.query(User).filter(User.email == req.email).first()
|
|
if existing:
|
|
raise AppException("Email already registered", 409)
|
|
|
|
# Find or create tenant
|
|
tenant = db.query(Tenant).filter(Tenant.slug == req.tenant_slug).first()
|
|
if not tenant:
|
|
if not req.tenant_name:
|
|
raise AppException(
|
|
f"Tenant '{req.tenant_slug}' not found. Provide tenant_name to create one.",
|
|
404,
|
|
)
|
|
tenant = Tenant(name=req.tenant_name, slug=req.tenant_slug)
|
|
db.add(tenant)
|
|
db.flush()
|
|
tenant_id = tenant.id
|
|
# First user in a new tenant becomes tenant_admin
|
|
role = "tenant_admin"
|
|
else:
|
|
tenant_id = tenant.id
|
|
role = req.role
|
|
|
|
user = User(
|
|
email=req.email,
|
|
username=req.username,
|
|
hashed_password=hash_password(req.password),
|
|
role=role,
|
|
user_type="human",
|
|
tenant_id=tenant_id,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
token = create_access_token(
|
|
subject=str(user.id),
|
|
tenant_id=user.tenant_id,
|
|
role=user.role,
|
|
)
|
|
return TokenResponse(
|
|
access_token=token,
|
|
user=UserBrief.model_validate(user),
|
|
)
|
|
|
|
@staticmethod
|
|
def get_current_user(db: Session, user_id: int) -> User:
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise NotFoundError("User", user_id)
|
|
return user
|