"""Pydantic-схемы админ-API (/api/v1/admin, гейт require_admin)."""
import uuid
from datetime import datetime
from pydantic import BaseModel, Field
SlugPattern = r"^[a-z0-9]([a-z0-9._-]*[a-z0-9])?$"
# --- sources / keys ---
class SourceIn(BaseModel):
name: str = Field(min_length=2, max_length=64, pattern=SlugPattern)
label: str | None = Field(None, max_length=120)
description: str | None = None
class SourceOut(BaseModel):
id: int
name: str
label: str | None
description: str | None
created_at: datetime
keys_count: int = 0
types_count: int = 0
class KeyIn(BaseModel):
name: str = Field(min_length=1, max_length=120)
class KeyOut(BaseModel):
id: int
name: str
token_hint: str
created_at: datetime
last_used_at: datetime | None
revoked_at: datetime | None
class KeyCreated(KeyOut):
"""Токен печатается/показывается МИНУС один раз — только в этом ответе."""
token: str
# --- types ---
class TypeIn(BaseModel):
source_id: int
subject: str = Field(min_length=1, max_length=64, pattern=SlugPattern)
action: str = Field(min_length=1, max_length=64, pattern=SlugPattern)
payload_schema: dict | None = None
description: str | None = None
class TypeOut(TypeIn):
id: int
source_name: str
created_at: datetime
# --- targets ---
class TargetIn(BaseModel):
channel: str = Field(pattern=r"^(telegram|email|s2s|internal_log)$")
name: str = Field(min_length=1, max_length=120)
description: str | None = None
config: dict = {}
class TargetOut(TargetIn):
id: int
enabled: bool
created_at: datetime
class TargetPatch(BaseModel):
"""Частичное обновление цели; канал не меняется (ссылки из правил)."""
name: str | None = Field(default=None, min_length=1, max_length=120)
description: str | None = None
config: dict | None = None
enabled: bool | None = None
# --- rules ---
class RuleActionIn(BaseModel):
channel: str = Field(pattern=r"^(telegram|email|s2s|internal_log|user)$")
target_id: int | None = None
template: str | None = None
class RuleActionOut(RuleActionIn):
id: int
target_name: str | None = None
class RuleIn(BaseModel):
name: str = Field(min_length=1, max_length=120)
conditions: dict = {}
template: str | None = None
throttle_seconds: int | None = Field(None, ge=1)
weight: int = 0
actions: list[RuleActionIn] = Field(min_length=1)
class RuleOut(RuleIn):
id: int
enabled: bool
created_at: datetime
actions: list[RuleActionOut]
class RulePatch(BaseModel):
name: str | None = Field(None, max_length=120)
enabled: bool | None = None
template: str | None = None
throttle_seconds: int | None = Field(None, ge=1)
weight: int | None = None
conditions: dict | None = None
actions: list[RuleActionIn] | None = None
# --- events / deliveries ---
class DeliveryOut(BaseModel):
id: int
event_id: uuid.UUID
rule_id: int | None
channel: str
target_id: int | None
target_name: str | None = None
# Канал user — кому из пользователей доставлено (sub gnexus-auth).
recipient_user_id: str | None = None
status: str
attempts: int
last_error: str | None
next_retry_at: datetime | None
rendered_message: str | None
created_at: datetime
delivered_at: datetime | None
class AdminEventOut(BaseModel):
id: uuid.UUID
source_id: int
source_name: str
subject: str
action: str
priority: str
status: str
payload: dict
dedup_key: str | None
created_at: datetime
expires_at: datetime | None
deliveries: list[DeliveryOut]