saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Self
|
|
|
|
from email_validator import EmailNotValidError, validate_email
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
from app.models.enums import NotificationType, SeverityLevel
|
|
from app.models.notification_destination import NotificationDestination
|
|
|
|
_PHONE_RE = re.compile(r"^\+?[1-9]\d{7,14}$")
|
|
|
|
|
|
class NotificationDestinationCreate(BaseModel):
|
|
type: NotificationType
|
|
destination_value: str = Field(min_length=1, max_length=320)
|
|
minimum_severity: SeverityLevel = SeverityLevel.MEDIUM
|
|
enabled: bool = True
|
|
company_ids: list[uuid.UUID] = Field(
|
|
min_length=1,
|
|
description="Which companies this destination receives alerts for. If a destination "
|
|
"with the same type/value already exists for this user, it's reused (linked to these "
|
|
"companies too) rather than duplicated.",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_destination_value(self) -> Self:
|
|
if self.type == NotificationType.EMAIL:
|
|
try:
|
|
validate_email(self.destination_value, check_deliverability=False)
|
|
except EmailNotValidError as exc:
|
|
raise ValueError(f"Invalid email address: {exc}") from exc
|
|
elif self.type == NotificationType.SMS:
|
|
if not _PHONE_RE.match(self.destination_value):
|
|
raise ValueError("Phone number must be in E.164 format, e.g. +15551234567")
|
|
return self
|
|
|
|
|
|
class NotificationDestinationUpdate(BaseModel):
|
|
destination_value: str | None = Field(default=None, min_length=1, max_length=320)
|
|
enabled: bool | None = None
|
|
minimum_severity: SeverityLevel | None = None
|
|
|
|
|
|
class LinkedCompany(BaseModel):
|
|
id: uuid.UUID
|
|
name: str
|
|
|
|
|
|
class NotificationDestinationResponse(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: uuid.UUID
|
|
type: NotificationType
|
|
destination_value: str
|
|
verified: bool
|
|
enabled: bool
|
|
minimum_severity: SeverityLevel
|
|
created_at: datetime
|
|
companies: list[LinkedCompany] = Field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_destination(cls, destination: NotificationDestination) -> Self:
|
|
return cls(
|
|
id=destination.id,
|
|
type=destination.type,
|
|
destination_value=destination.destination_value,
|
|
verified=destination.verified,
|
|
enabled=destination.enabled,
|
|
minimum_severity=destination.minimum_severity,
|
|
created_at=destination.created_at,
|
|
companies=[
|
|
LinkedCompany(id=link.company.id, name=link.company.name)
|
|
for link in destination.company_links
|
|
],
|
|
)
|
|
|
|
|
|
class NotificationTestResult(BaseModel):
|
|
success: bool
|
|
error: str | None = None
|