38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
import uuid
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from domains.notifications.models import Notification
|
|
|
|
async def create_notification(db: AsyncSession, notification: Notification):
|
|
db.add(notification)
|
|
await db.commit()
|
|
await db.refresh(notification)
|
|
return notification
|
|
|
|
async def get_notification_by_id(db: AsyncSession, notification_id: uuid.UUID):
|
|
result = await db.execute(
|
|
select(Notification).where(Notification.id == notification_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_user_notifications(db: AsyncSession, user_id: uuid.UUID):
|
|
result = await db.execute(
|
|
select(Notification)
|
|
.where(Notification.receiver_id == user_id)
|
|
.order_by(Notification.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_all_notifications(db: AsyncSession):
|
|
result = await db.execute(
|
|
select(Notification)
|
|
.order_by(Notification.created_at.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def update_notification(db: AsyncSession, notification: Notification):
|
|
await db.commit()
|
|
await db.refresh(notification)
|
|
return notification
|