To count notifications that wasn't read by a user, I use the following code:
query = query.filter(
or_(Notifs.tid == current_user.id, # is targeted to the user
Notifs.target_groups.any(Role.id.in_([r.id for r in current_user.roles])))) # user is among the target groups of the notification
count = 0
for notif in query.all():
q1 = UserReads.query.filter_by(uid = current_user.id, nid = notif.id).first()
if not q1 or not q1.read:
count += 1
Could it be done in a single query and then using query.count()?
The "Option 2" below I think is the most clear and I think the list of "read" notifications will be shorter than unread because it will be filtered explicitly by the user. I haven't found great docs on using "any_".
This should work for SQLAlchemy 2.0.
import sys
from sqlalchemy import (
create_engine,
Integer,
DateTime,
Boolean,
)
from sqlalchemy.schema import (
Column,
ForeignKey,
UniqueConstraint,
Table,
)
from sqlalchemy.sql import select, func, and_, or_, not_
from sqlalchemy.orm import declarative_base, Session, aliased, relationship
Base = declarative_base()
username, password, db = sys.argv[1:4]
engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=True)
class Notification(Base):
__tablename__ = "notifications"
id = Column(Integer, primary_key=True)
tid = Column(Integer, ForeignKey('users.id'), nullable=True)
target_roles = relationship("Role", back_populates='notifications', secondary='notifications_roles')
user_reads = relationship("UserRead", back_populates='notification')
target_user = relationship("User")
class Role(Base):
__tablename__ = 'roles'
id = Column(Integer, primary_key=True)
users = relationship("User", secondary="users_roles", back_populates="roles")
notifications = relationship("Notification", back_populates='target_roles', secondary='notifications_roles')
notifications_roles = Table('notifications_roles', Base.metadata,
Column("id", Integer, primary_key=True),
Column("role_id", Integer, ForeignKey('roles.id'), nullable=False),
Column("notification_id", Integer, ForeignKey('notifications.id'), nullable=False),
UniqueConstraint("notification_id", "role_id", name="notification_role_uc"))
users_roles_t = Table("users_roles", Base.metadata,
Column("id", Integer, primary_key=True),
Column("user_id", ForeignKey('users.id'), nullable=False),
Column("role_id", ForeignKey('roles.id'), nullable=False),
UniqueConstraint("user_id", "role_id", name="user_role_uc"),
)
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
roles = relationship("Role", secondary="users_roles", back_populates="users")
user_reads = relationship("UserRead", back_populates='user')
class UserRead(Base):
__tablename__ = 'user_reads'
id = Column(Integer, primary_key=True)
uid = Column(Integer, ForeignKey('users.id'), nullable=False)
nid = Column(Integer, ForeignKey('notifications.id'), nullable=False)
read = Column(Boolean, nullable=False, default=True)
__table_args__ = (
UniqueConstraint('uid', 'nid'),
)
user = relationship(User, back_populates='user_reads')
notification = relationship(Notification, back_populates='user_reads')
Base.metadata.create_all(engine)
with Session(engine) as session:
"""
Set up 3 notification to demonstrate different unread/read states.
"""
u1 = User()
session.add(u1)
session.flush()
r1 = Role()
r2 = Role()
session.add_all([r1, r2])
session.flush()
u1.roles.append(r1)
session.flush()
# Notifications target user.
n1, n2, n3 = Notification(), Notification(), Notification()
session.add_all([n1, n2, n3])
n1.tid = n2.tid = n3.tid = u1.id
session.flush()
#First case: User read exists and it is False. (should count)
ur1 = UserRead(read=False)
session.add(ur1)
ur1.notification = n1
ur1.user = u1
#Second case: User read exists but it is True (should not count)
ur2 = UserRead(read=True)
session.add(ur2)
ur2.notification = n2
ur2.user = u1
session.flush()
session.commit()
#Third case: User read does not exist (should count)
#...
with Session(engine) as session:
"""
Setup 4 cases, the first 3 are similar to the above cases but notification is targeted at user's role instead of user.
The last case does not target the user at all so should not be counted as unread even though no UserRead exists.
"""
u1 = session.query(User).first()
r1 = u1.roles[0]
r2 = session.query(Role).where(Role.id != r1.id).first()
n1, n2, n3, n4 = Notification(), Notification(), Notification(), Notification()
session.add_all([n1, n2, n3, n4])
# Target u1's role.
n1.target_roles.append(r1)
n2.target_roles.append(r1)
n3.target_roles.append(r1)
# Does not target u1's role.
n4.target_roles.append(r2)
session.flush()
#First case: UserRead exists and it is False. (should count)
ur1 = UserRead(read=False)
session.add(ur1)
ur1.notification = n1
ur1.user = u1
#Second case: UserRead exists but it is True (should not count)
ur2 = UserRead(read=True)
session.add(ur2)
ur2.notification = n2
ur2.user = u1
session.flush()
session.commit()
#Third case: UserRead does not exist (should count)
#...
#Fourth case: User's role is not targeted (should not count)
#...
with Session(engine) as session:
current_user = session.query(User).first()
#
# Option 1, using unread subquery.
#
target_subq = select(Notification.id).outerjoin(
Notification.target_roles
).where(
or_(
Notification.tid == current_user.id,
Role.id.in_([r.id for r in current_user.roles])))
# Decide which notifications have not been read by this user.
unread_subq = select(Notification.id).outerjoin(
UserRead, and_(
Notification.id == UserRead.nid,
UserRead.uid == current_user.id
)).where(or_(
# We recorded current user didn't read it.
UserRead.read == False,
# There is no UserRead record at all for current user.
UserRead.id == None))
target_unread_q = select(
func.count(Notification.id)
).where(
Notification.id.in_(unread_subq),
Notification.id.in_(target_subq))
print (session.execute(target_unread_q).scalar())
#
# Option 2, using negation of read subquery.
#
read_subq = select(Notification.id).join(
UserRead, and_(
Notification.id == UserRead.nid,
UserRead.uid == current_user.id
)).where(
UserRead.read == True)
target_unread_q = select(
func.count(Notification.id)
).where(
# It has not been read.
Notification.id.not_in(read_subq),
# It matches the target.
Notification.id.in_(target_subq))
print (session.execute(target_unread_q).scalar())
The query format string (taken from log output of echo = True) for option 2.
"""SELECT count(notifications.id) AS count_1
FROM notifications
WHERE (notifications.id NOT IN (SELECT notifications.id
FROM notifications JOIN user_reads ON notifications.id = user_reads.nid AND user_reads.uid = %(uid_1)s
WHERE user_reads.read = true)) AND notifications.id IN (SELECT notifications.id
FROM notifications LEFT OUTER JOIN (notifications_roles AS notifications_roles_1 JOIN roles ON roles.id = notifications_roles_1.role_id) ON notifications.id = notifications_roles_1.notification_id
WHERE notifications.tid = %(tid_1)s OR roles.id IN (%(id_1_1)s))"""
The substitutions for the query:
{'uid_1': 1, 'tid_1': 1, 'id_1_1': 1}
This should return 4 matching "unread" notifications.