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

from datetime import datetime
from typing import TYPE_CHECKING, Generic, TypeVar

from core_shared import JSONValue
from pydantic import BaseModel, Field

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

T = TypeVar("T")


class ApiErrorResponse(BaseModel):
    errorType: str
    message: str
    details: dict[str, JSONValue] = Field(default_factory=dict)


class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T | None = None
    error: ApiErrorResponse | None = None
    requestId: str
    serverTime: datetime


class PageRequest(BaseModel):
    page: int = Field(default=1, ge=1)
    pageSize: int = Field(default=20, ge=1, le=200)
    keyword: str | None = None
    sortBy: str = "createdTime"
    sortOrder: str = "desc"

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.pageSize


class PageResult(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    pageSize: int
    hasMore: bool

    @classmethod
    def from_items(
        cls,
        *,
        items: list[T],
        total: int,
        page: int,
        page_size: int) -> "PageResult[T]":
        return cls(
            items=items,
            total=total,
            page=page,
            pageSize=page_size,
            hasMore=page * page_size < total)


class UserDto(BaseModel):
    id: str
    username: str
    displayName: str | None = None
    email: str | None = None
    metadata: dict[str, JSONValue] = Field(default_factory=dict)
    lastLoginTime: datetime | None = None
    createdTime: datetime
    updatedTime: datetime

    @classmethod
    def from_entity(cls, entity: "User") -> "UserDto":
        return cls(
            id=entity.id,
            username=entity.username,
            displayName=entity.display_name,
            email=entity.email,
            metadata=entity.metadata_json or {},
            lastLoginTime=entity.last_login_time,
            createdTime=entity.created_time,
            updatedTime=entity.updated_time)


class RoleDto(BaseModel):
    id: str
    name: str
    description: str | None = None
    permissionBindingCount: int
    createdTime: datetime
    updatedTime: datetime

    @classmethod
    def from_entity(cls, entity: "Role", *, permission_binding_count: int = 0) -> "RoleDto":
        return cls(
            id=entity.id,
            name=entity.name,
            description=entity.description,
            permissionBindingCount=permission_binding_count,
            createdTime=entity.created_time,
            updatedTime=entity.updated_time)


class RolePermissionBindingDto(BaseModel):
    id: str
    roleId: str
    permission: str
    scopeType: str | None = None
    scopeId: str | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "RolePermissionBinding") -> "RolePermissionBindingDto":
        return cls(
            id=entity.id,
            roleId=entity.role_id,
            permission=entity.permission,
            scopeType=entity.scope_type,
            scopeId=entity.scope_id,
            createdTime=entity.created_time)


class ApiKeyDto(BaseModel):
    id: str
    name: str
    keyPrefix: str
    scopes: str | None = None
    expiresTime: datetime | None = None
    lastUsedTime: datetime | None = None
    revokedTime: datetime | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "ApiKey") -> "ApiKeyDto":
        return cls(
            id=entity.id,
            name=entity.name,
            keyPrefix=entity.key_prefix,
            scopes=entity.scopes,
            expiresTime=entity.expires_time,
            lastUsedTime=entity.last_used_time,
            revokedTime=entity.revoked_time,
            createdTime=entity.created_time)


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


class LoginData(BaseModel):
    accessToken: str
    tokenType: str = "bearer"
    expiresTime: datetime
    user: UserDto


class TokenVerifyRequestDto(BaseModel):
    accessToken: str


class TokenVerifyData(BaseModel):
    active: bool
    userId: str | None = None
    username: str | None = None
    expiresTime: datetime | None = None
    reason: str | None = None


class AuthMeData(BaseModel):
    user: UserDto
    roles: list[RoleDto]
    permissions: list[str]


class RolePermissionBindingAddRequest(BaseModel):
    roleId: str
    permission: str
    scopeType: str | None = None
    scopeId: str | None = None


class RolePermissionBindingListRequest(PageRequest):
    roleId: str


class BindingRemoveRequest(BaseModel):
    bindingId: str


class PermissionCheckRequestDto(BaseModel):
    userId: str
    permission: str
    scopeType: str | None = None
    scopeId: str | None = None


class PermissionCheckData(BaseModel):
    allowed: bool
    reason: str
    matchedRoleIds: list[str] = Field(default_factory=list)


class ApiKeyCreateRequestDto(BaseModel):
    name: str
    scopes: str | None = None
    expiresTime: datetime | None = None


class ApiKeyCreateData(BaseModel):
    apiKey: ApiKeyDto
    secret: str


class ApiKeyRevokeRequest(BaseModel):
    apiKeyId: str


class DeleteData(BaseModel):
    deleted: bool
    bindingId: str | None = None
    apiKeyId: str | None = None