55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
from sqlalchemy import INTEGER, DateTime, Enum, ForeignKey, UniqueConstraint
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from db.base import Base
|
|
from schemas.enums import NotificationStatus, NotificationType
|
|
|
|
if TYPE_CHECKING:
|
|
from db.models import Subscription
|
|
|
|
|
|
class ServiceNotification(Base):
|
|
__tablename__ = "service_notifications"
|
|
|
|
id: Mapped[int] = mapped_column(
|
|
INTEGER, unique=True, autoincrement=True, primary_key=True, nullable=False
|
|
)
|
|
subscription_id: Mapped[int] = mapped_column(
|
|
ForeignKey("subscriptions.id"), nullable=False, unique=False
|
|
)
|
|
notify_type: Mapped[NotificationType] = mapped_column(
|
|
Enum(NotificationType, name="notificationtype"), nullable=False
|
|
)
|
|
sub_expires_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), unique=True, nullable=False
|
|
)
|
|
|
|
status: Mapped[NotificationStatus] = mapped_column(
|
|
Enum(NotificationStatus, name="notificationstatus"),
|
|
nullable=False,
|
|
default=NotificationStatus.PENDING,
|
|
)
|
|
dispatched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
sent_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
attempts: Mapped[int] = mapped_column(INTEGER, nullable=False, default=0)
|
|
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
default=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
__table_args__ = (
|
|
UniqueConstraint(
|
|
"subscription_id",
|
|
"notify_type",
|
|
"sub_expires_at",
|
|
name="uq_subscription_notification",
|
|
),
|
|
)
|
|
|
|
subscription: Mapped["Subscription"] = relationship("Subscription", lazy="selectin")
|