65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
from sqlalchemy import func, or_, select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.service_notifications import ServiceNotification
|
|
from schemas.enums import NotificationStatus
|
|
|
|
|
|
async def get_notification_by_id(session: AsyncSession, n_id: int) -> ServiceNotification | None:
|
|
stmt = select(ServiceNotification).where(ServiceNotification.id == n_id)
|
|
r = await session.execute(stmt)
|
|
|
|
return r.scalar_one_or_none()
|
|
|
|
|
|
async def get_pending_notifications(
|
|
session: AsyncSession, batch_size: int = 50
|
|
) -> list[ServiceNotification]:
|
|
stmt = (
|
|
select(ServiceNotification)
|
|
.where(
|
|
or_(
|
|
ServiceNotification.status == "pending",
|
|
(
|
|
(ServiceNotification.status == "dispatched")
|
|
& (
|
|
ServiceNotification.dispatched_at
|
|
< func.now() - text("interval '10 minutes'")
|
|
)
|
|
),
|
|
)
|
|
)
|
|
.order_by(ServiceNotification.created_at)
|
|
.with_for_update(skip_locked=True)
|
|
.limit(batch_size)
|
|
)
|
|
r = await session.execute(stmt)
|
|
|
|
return list(r.scalars().all())
|
|
|
|
|
|
async def ack_notification(session: AsyncSession, n_id: int) -> ServiceNotification | None:
|
|
notification = await get_notification_by_id(session, n_id)
|
|
if not notification:
|
|
return
|
|
|
|
notification.sent_at = func.now()
|
|
notification.status = NotificationStatus.SENT
|
|
|
|
await session.commit()
|
|
|
|
return notification
|
|
|
|
|
|
async def mark_notification_as_dispatched(session: AsyncSession, n_id: int):
|
|
notification = await get_notification_by_id(session, n_id)
|
|
if not notification:
|
|
return
|
|
|
|
notification.dispatched_at = func.now()
|
|
notification.status = NotificationStatus.DISPATCHED
|
|
notification.attempts += 1
|
|
|
|
await session.commit()
|
|
return notification
|