discours.io/notifier
5
0
Fork 0
Code Pull Requests Actions Activity

after-fix
All checks were successful
deploy / deploy (push) Successful in 1m7s
Details

Browse Source
This commit is contained in:
Untone 2023-12-18 18:41:46 +03:00
parent 25b6a6f50a
commit 5c6a680832
1 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
resolvers/schema.py
View File

@ -1,5 +1,6 @@
import logging
from typing import List
from sqlalchemy.schema import Column
import strawberry
from sqlalchemy import and_, select
@ -42,7 +43,7 @@ class NotificationsResult:
total: int
def get_notifications(author_id: int, session, after: int, limit: int = 9999, offset: int = 0) -> List[Notification]:
def get_notifications(author_id: int, session, after: int | Column[int], limit: int = 9999, offset: int = 0) -> List[Notification]:
NotificationSeenAlias = aliased(NotificationSeen)
query = (
select(NotificationMessage, NotificationSeenAlias.viewer.label("seen"))
@ -76,13 +77,11 @@ def get_notifications(author_id: int, session, after: int, limit: int = 9999, of
@strawberry.type
class Query:
@strawberry.field
async def load_notifications(self, info, limit: int = 50, offset: int = 0) -> NotificationsResult:
async def load_notifications(self, info, after: int, limit: int = 50, offset: int = 0) -> NotificationsResult:
author_id = info.context.get("author_id")
with local_session() as session:
try:
if author_id:
author = session.query(Author).filter(Author.id == author_id).first()
after = author.last_seen
notifications = get_notifications(author_id, session, after, limit, offset)
if notifications and len(notifications) > 0:
nr = NotificationsResult(
@ -124,12 +123,15 @@ class Mutation:
if author_id:
with local_session() as session:
try:
nslist = get_notifications(author_id, session)
for n in nslist:
if author_id not in n.seen:
ns = NotificationSeen(viewer=author_id, notification=n.id)
session.add(ns)
session.commit()
author = session.query(Author).filter(Author.id == author_id).first()
if author:
after = author.last_seen
nslist = get_notifications(author_id, session, after)
for n in nslist:
if author_id not in n.seen:
ns = NotificationSeen(viewer=author_id, notification=n.id)
session.add(ns)
session.commit()
except SQLAlchemyError as e:
session.rollback()
logger.error(