from __future__ import annotations from datetime import datetime from typing import Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from app.security.scopes import ALL_SCOPES Environment = Literal["live", "test"] class APIKeyCreate(BaseModel): name: str = Field(min_length=1, max_length=120) environment: Environment = "live" role: str | None = Field(default="viewer", min_length=1, max_length=64) scopes: list[str] = Field(default_factory=list) expires_at: datetime | None = None expires_in_seconds: int | None = Field(default=None, ge=60, le=31_536_000) notes: str | None = Field(default=None, max_length=4000) requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000) concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000) uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000) processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576) @field_validator("name") @classmethod def validate_name(cls, value: str) -> str: normalized = value.strip() if not normalized: raise ValueError("API key name cannot be blank") return normalized @field_validator("scopes") @classmethod def validate_scopes(cls, values: list[str]) -> list[str]: normalized = list(dict.fromkeys(value.strip().lower() for value in values)) unknown = set(normalized) - ALL_SCOPES if unknown: raise ValueError(f"Unsupported scopes: {sorted(unknown)}") return normalized @model_validator(mode="after") def validate_expiration(self) -> APIKeyCreate: if self.expires_at is not None and self.expires_in_seconds is not None: raise ValueError("Use either expires_at or expires_in_seconds, not both") return self class APIKeyPatch(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=120) role: str | None = Field(default=None, min_length=1, max_length=64) scopes: list[str] | None = None expires_at: datetime | None = None clear_expiration: bool = False notes: str | None = Field(default=None, max_length=4000) requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000) concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000) uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000) processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576) @field_validator("name") @classmethod def validate_name(cls, value: str | None) -> str | None: if value is None: return None normalized = value.strip() if not normalized: raise ValueError("API key name cannot be blank") return normalized @field_validator("scopes") @classmethod def validate_scopes(cls, values: list[str] | None) -> list[str] | None: if values is None: return None normalized = list(dict.fromkeys(value.strip().lower() for value in values)) unknown = set(normalized) - ALL_SCOPES if unknown: raise ValueError(f"Unsupported scopes: {sorted(unknown)}") return normalized @model_validator(mode="after") def validate_expiration(self) -> APIKeyPatch: if self.clear_expiration and "expires_at" in self.model_fields_set: raise ValueError("Use either expires_at or clear_expiration, not both") return self class APIKeyRotate(BaseModel): grace_period_seconds: int = Field(default=0, ge=0, le=86_400) class APIKeyView(BaseModel): model_config = ConfigDict(from_attributes=True) id: str name: str key_prefix: str environment: str status: str role: str | None scopes: list[str] created_at: datetime last_used_at: datetime | None expires_at: datetime | None grace_expires_at: datetime | None created_by: str | None notes: str | None rotated_from_id: str | None requests_per_minute: int concurrent_jobs: int uploads_per_hour: int processing_bytes_per_day: int class APIKeyCreated(APIKeyView): api_key: str class APIKeyList(BaseModel): items: list[APIKeyView] total: int class AuthContextView(BaseModel): id: str name: str key_prefix: str environment: str role: str | None scopes: list[str] expires_at: datetime | None workspace_id: str | None = None user_id: str | None = None membership_role: str | None = None class AuditLogView(BaseModel): model_config = ConfigDict(from_attributes=True) id: str request_id: str api_key_id: str | None key_name: str | None ip_address: str | None user_agent: str | None endpoint: str http_method: str response_code: int processing_time_ms: int bytes_uploaded: int bytes_downloaded: int created_at: datetime