"""Pydantic schemas for the identity-consumer self-service request flow.""" from __future__ import annotations import re from datetime import datetime from typing import Any from uuid import UUID from pydantic import BaseModel, Field, field_validator from app.identity.consumers.schemas import _validate_scopes _SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$") class ConsumerRequestCreate(BaseModel): """Self-service application body. The applicant is the authenticated caller; ``owner_person_id`` defaults to them when omitted.""" name: str = Field(min_length=1, max_length=128) slug: str = Field(min_length=2, max_length=64) description: str | None = Field(default=None, max_length=2000) owner_person_id: UUID | None = None scopes: list[str] granted_fields: list[str] | None = None filter_config: dict[str, Any] = Field(default_factory=dict) rate_limit_per_minute: int | None = Field(default=None, ge=1, le=10000) #: Token-expiry preset in days; null = never expires. token_validity_days: int | None = Field(default=None, ge=1, le=3650) pii_justification: str | None = Field(default=None, max_length=2000) @field_validator("slug") @classmethod def _slug_shape(cls, v: str) -> str: v = v.strip().lower() if not _SLUG_RE.match(v): raise ValueError( "slug must be lowercase letters/digits/dashes, " "2-64 chars, no leading/trailing dash" ) return v @field_validator("scopes") @classmethod def _scopes_known(cls, v: list[str]) -> list[str]: return _validate_scopes(v) @field_validator("token_validity_days") @classmethod def _validity_preset(cls, v: int | None) -> int | None: # Must be one of the platform's token-expiry presets (null = # never). Rejecting an off-preset value here — not at the terminal # one-time issue step, where ``mint`` would raise and strand the # consumer at zero tokens on an already-spent link. if v is None: return v from app.identity.consumers.token_service import ALLOWED_EXPIRY_DAYS if v not in ALLOWED_EXPIRY_DAYS: raise ValueError( f"token_validity_days must be one of {sorted(ALLOWED_EXPIRY_DAYS)} " "or null (never expires)" ) return v class ConsumerRequestResponse(BaseModel): id: UUID requester_person_id: UUID requester_display_name: str | None = None name: str slug: str description: str | None owner_person_id: UUID owner_display_name: str | None = None scopes: list[str] granted_fields: list[str] filter_config: dict[str, Any] rate_limit_per_minute: int | None token_validity_days: int | None pii_justification: str | None status: str reject_reason: str | None feishu_instance_code: str | None consumer_id: UUID | None decided_at: datetime | None created_at: datetime updated_at: datetime model_config = {"from_attributes": True} __all__ = ["ConsumerRequestCreate", "ConsumerRequestResponse"]