agent-platform/services/auth-service/app/schemas/auth.py

from datetime import datetime
from typing import TYPE_CHECKING

from core_domain import (
    PermissionCheckContract,
    PermissionCheckResultContract,
    RoleAssignmentContract,
    RoleAssignmentStatus,
    RoleContract,
    RoleStatus,
    UserContract,
    UserStatus,
)
from core_shared import JSONValue
from pydantic import BaseModel, Field

if TYPE_CHECKING:
    from app.db.models import Role, RoleAssignment, User


class UserCreateRequest(BaseModel):
    username: str
    password: str | None = Field(default=None, min_length=8)
    display_name: str | None = None
    email: str | None = None
    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)


class UserStatusUpdateRequest(BaseModel):
    status: UserStatus


class UserResponse(UserContract):
    @classmethod
    def from_entity(cls, entity: "User") -> "UserResponse":
        return cls.model_validate(entity, from_attributes=True)


class LoginUserResponse(BaseModel):
    id: str
    username: str
    display_name: str | None = None
    email: str | None = None
    status: UserStatus
    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)
    last_login_time: datetime | None = None
    created_time: datetime

    @classmethod
    def from_entity(cls, entity: "User") -> "LoginUserResponse":
        return cls.model_validate(entity, from_attributes=True)


class LoginRequest(BaseModel):
    username: str
    password: str


class LoginResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
    expires_time: datetime
    user: LoginUserResponse


class TokenVerifyRequest(BaseModel):
    access_token: str


class TokenVerifyResponse(BaseModel):
    active: bool
    user_id: str | None = None
    username: str | None = None
    expires_time: datetime | None = None
    reason: str | None = None


class RoleCreateRequest(BaseModel):
    code: str
    name: str
    description: str | None = None
    permissions_json: list[str] = Field(default_factory=list)


class RoleStatusUpdateRequest(BaseModel):
    status: RoleStatus


class RoleResponse(RoleContract):
    @classmethod
    def from_entity(cls, entity: "Role") -> "RoleResponse":
        return cls.model_validate(entity, from_attributes=True)


class RoleAssignmentCreateRequest(BaseModel):
    user_id: str
    role_id: str
    scope_type: str | None = None
    scope_id: str | None = None
    expires_time: datetime | None = None


class RoleAssignmentStatusUpdateRequest(BaseModel):
    status: RoleAssignmentStatus


class RoleAssignmentResponse(RoleAssignmentContract):
    @classmethod
    def from_entity(cls, entity: "RoleAssignment") -> "RoleAssignmentResponse":
        return cls.model_validate(entity, from_attributes=True)


class PermissionCheckRequest(PermissionCheckContract):
    pass


class PermissionCheckResponse(PermissionCheckResultContract):
    pass