mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
129 lines
3.3 KiB
Python
129 lines
3.3 KiB
Python
"""
|
|
Schema Pydantic per i canali di notifica multi-canale.
|
|
"""
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
# ─── NotificationChannel ─────────────────────────────────────────────────────
|
|
|
|
ChannelType = Literal["webhook", "email", "telegram", "whatsapp"]
|
|
|
|
|
|
class NotificationChannelCreate(BaseModel):
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
channel_type: ChannelType
|
|
config: dict[str, Any] | None = None # configurazione pubblica
|
|
config_secret: dict[str, Any] | None = None # verrà cifrato in config_enc
|
|
|
|
|
|
class NotificationChannelUpdate(BaseModel):
|
|
name: str | None = None
|
|
is_active: bool | None = None
|
|
config: dict[str, Any] | None = None
|
|
config_secret: dict[str, Any] | None = None
|
|
|
|
|
|
class NotificationChannelResponse(BaseModel):
|
|
id: uuid.UUID
|
|
tenant_id: uuid.UUID
|
|
name: str
|
|
channel_type: str
|
|
is_active: bool
|
|
config: dict[str, Any] | None
|
|
consecutive_failures: int
|
|
circuit_open_until: datetime | None
|
|
created_by: uuid.UUID | None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class NotificationChannelListResponse(BaseModel):
|
|
items: list[NotificationChannelResponse]
|
|
total: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
class ChannelTestResult(BaseModel):
|
|
success: bool
|
|
message: str
|
|
http_status: int | None = None
|
|
|
|
|
|
# ─── NotificationRule ─────────────────────────────────────────────────────────
|
|
|
|
EventType = Literal[
|
|
"new_message",
|
|
"state_changed",
|
|
"anomaly",
|
|
"send_failed",
|
|
"send_delivered",
|
|
"mailbox_error",
|
|
]
|
|
|
|
|
|
class NotificationRuleCreate(BaseModel):
|
|
channel_id: uuid.UUID
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
event_type: EventType
|
|
filter: dict[str, Any] | None = None
|
|
|
|
|
|
class NotificationRuleUpdate(BaseModel):
|
|
name: str | None = None
|
|
event_type: EventType | None = None
|
|
filter: dict[str, Any] | None = None
|
|
is_active: bool | None = None
|
|
|
|
|
|
class NotificationRuleResponse(BaseModel):
|
|
id: uuid.UUID
|
|
tenant_id: uuid.UUID
|
|
channel_id: uuid.UUID
|
|
name: str
|
|
event_type: str
|
|
filter: dict[str, Any] | None
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class NotificationRuleListResponse(BaseModel):
|
|
items: list[NotificationRuleResponse]
|
|
total: int
|
|
|
|
|
|
# ─── NotificationLog ─────────────────────────────────────────────────────────
|
|
|
|
class NotificationLogResponse(BaseModel):
|
|
id: uuid.UUID
|
|
tenant_id: uuid.UUID
|
|
channel_id: uuid.UUID
|
|
rule_id: uuid.UUID | None
|
|
event_type: str
|
|
status: str
|
|
attempt_count: int
|
|
max_attempts: int
|
|
next_retry_at: datetime | None
|
|
last_error: str | None
|
|
http_status: int | None
|
|
sent_at: datetime | None
|
|
created_at: datetime
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class NotificationLogListResponse(BaseModel):
|
|
items: list[NotificationLogResponse]
|
|
total: int
|
|
page: int
|
|
page_size: int
|