19325323565

Committed 13 Nov 2025 08:29AM UTC coverage: 86.236% (-0.1%) from 86.355%

Build # 19325323565

Build Type

Pull #1094

github

Committed by

web-flow

Commit Message

Merge 371b823e7 into 0ea39ae9f

Pull Request Pull Request #1094: feat: add user alerts

Run Details

244 of 311 new or added lines in 11 files covered. (78.46%)

104 existing lines in 5 files now uncovered.

23050 of 26729 relevant lines covered (86.24%)

1.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.56

/components/renku_data_services/notifications/db.py

"""Adapters for notification database classes."""

from collections.abc import Callable
from datetime import UTC, datetime

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ulid import ULID

from renku_data_services import base_models, errors
from renku_data_services.notifications import models
from renku_data_services.notifications import orm as schemas


class NotificationsRepository:
    """Repository for Notifications."""

    def __init__(
        self,
        session_maker: Callable[..., AsyncSession],
    ):
        self.session_maker = session_maker

    async def create_alert(self, user: base_models.APIUser, alert: models.UnsavedAlert) -> models.Alert:
        """Insert a new alert into the database."""
        if user.id is None:
            raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")
        if not user.is_admin:
            raise errors.ForbiddenError(message="You do not have the required permissions for this operation.")

        async with self.session_maker() as session:
            query = (
                select(schemas.AlertORM)
                .where(schemas.AlertORM.user_id == alert.user_id)
                .where(schemas.AlertORM.title == alert.title)
                .where(schemas.AlertORM.message == alert.message)
                .where(schemas.AlertORM.session_name == alert.session_name)
                .where(schemas.AlertORM.resolved_at.is_(None))
            )

            res = await session.scalars(query)
            existing_alert = res.one_or_none()
            if existing_alert is not None:
                raise errors.ConflictError(message="An identical unresolved alert already exists.")

        async with self.session_maker() as session, session.begin():
            alert_orm = schemas.AlertORM(
                title=alert.title,
                message=alert.message,
                user_id=alert.user_id,
                session_name=alert.session_name,
            )
            session.add(alert_orm)
            await session.flush()
            await session.refresh(alert_orm)
            return alert_orm.dump()

    async def get_alerts_for_user(
        self, user: base_models.APIUser, session_name: str | None = None
    ) -> list[models.Alert]:
        """Get all alerts for a given user."""
        if user.id is None:
            raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")

        async with self.session_maker() as session:
            query = (
                select(schemas.AlertORM)
                .where(schemas.AlertORM.user_id == user.id)
                .where(schemas.AlertORM.resolved_at.is_(None))
            )

            if session_name is not None:
                query = query.where(schemas.AlertORM.session_name == session_name)

            query = query.order_by(schemas.AlertORM.id.desc())
            alerts = await session.scalars(query)
            alert_list = alerts.all()
            return [alert.dump() for alert in alert_list]

    async def update_alert(self, user: base_models.APIUser, alert_id: ULID, patch: models.AlertPatch) -> models.Alert:
        """Update an existing alert in the database."""
        if user.id is None:
            raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")
        if not user.is_admin:
            raise errors.ForbiddenError(message="You do not have the required permissions for this operation.")

        async with self.session_maker() as session, session.begin():
            res = await session.scalars(select(schemas.AlertORM).where(schemas.AlertORM.id == alert_id))
            alert_orm = res.one_or_none()
            if alert_orm is None:
                raise errors.MissingResourceError(message=f"Alert with id '{alert_id}' not found.")

            self.__update_alert(alert_orm, patch)
            return alert_orm.dump()

    def __update_alert(self, alert: schemas.AlertORM, update: models.AlertPatch) -> None:
        if update.resolved is True:
            alert.resolved_at = datetime.now(UTC)

    async def get_alerts_by_properties(
        self,
        user: base_models.APIUser,
        alert_id: ULID | None,
        user_id: str | None,
        title: str | None,
        message: str | None,
        session_name: str | None,
        created_at: datetime | None,
        resolved_at: datetime | None,
    ) -> list[models.Alert]:
        """Get alerts by their properties."""
        if user.id is None:
            raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")

        async with self.session_maker() as session:
            query = select(schemas.AlertORM)

            if alert_id is not None:
                query = query.where(schemas.AlertORM.id == alert_id)

            if created_at is not None:
                query = query.where(schemas.AlertORM.creation_date == created_at)

            if title is None:
                query = query.where(schemas.AlertORM.title.is_(None))
            else:
                query = query.where(schemas.AlertORM.title == title)

            if message is None:
                query = query.where(schemas.AlertORM.message.is_(None))
            else:
                query = query.where(schemas.AlertORM.message == message)

            if resolved_at is None:
                query = query.where(schemas.AlertORM.resolved_at.is_(None))
            else:
                query = query.where(schemas.AlertORM.resolved_at == resolved_at)

            if session_name is None:
                query = query.where(schemas.AlertORM.session_name.is_(None))
            else:
                query = query.where(schemas.AlertORM.session_name == session_name)

            if user_id is None:
                query = query.where(schemas.AlertORM.user_id.is_(None))
            else:
                query = query.where(schemas.AlertORM.user_id == user_id)

            res = await session.scalars(query)
            alert_list = res.all()
            return [alert.dump() for alert in alert_list]

1	"""Adapters for notification database classes."""
2
3	from collections.abc import Callable	2✔
4	from datetime import UTC, datetime	2✔
5
6	from sqlalchemy import select	2✔
7	from sqlalchemy.ext.asyncio import AsyncSession	2✔
8	from ulid import ULID	2✔
9
10	from renku_data_services import base_models, errors	2✔
11	from renku_data_services.notifications import models	2✔
12	from renku_data_services.notifications import orm as schemas	2✔
13
14
15	class NotificationsRepository:	2✔
16	"""Repository for Notifications."""
17
18	def __init__(	2✔
19	self,
20	session_maker: Callable[..., AsyncSession],
21	):
22	self.session_maker = session_maker	2✔
23
24	async def create_alert(self, user: base_models.APIUser, alert: models.UnsavedAlert) -> models.Alert:	2✔
25	"""Insert a new alert into the database."""
26	if user.id is None:	1✔
NEW 27	raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")	×
28	if not user.is_admin:	1✔
NEW 29	raise errors.ForbiddenError(message="You do not have the required permissions for this operation.")	×
30
31	async with self.session_maker() as session:	1✔
32	query = (	1✔
33	select(schemas.AlertORM)
34	.where(schemas.AlertORM.user_id == alert.user_id)
35	.where(schemas.AlertORM.title == alert.title)
36	.where(schemas.AlertORM.message == alert.message)
37	.where(schemas.AlertORM.session_name == alert.session_name)
38	.where(schemas.AlertORM.resolved_at.is_(None))
39	)
40
41	res = await session.scalars(query)	1✔
42	existing_alert = res.one_or_none()	1✔
43	if existing_alert is not None:	1✔
NEW 44	raise errors.ConflictError(message="An identical unresolved alert already exists.")	×
45
46	async with self.session_maker() as session, session.begin():	1✔
47	alert_orm = schemas.AlertORM(	1✔
48	title=alert.title,
49	message=alert.message,
50	user_id=alert.user_id,
51	session_name=alert.session_name,
52	)
53	session.add(alert_orm)	1✔
54	await session.flush()	1✔
55	await session.refresh(alert_orm)	1✔
56	return alert_orm.dump()	1✔
57
58	async def get_alerts_for_user(	2✔
59	self, user: base_models.APIUser, session_name: str \| None = None
60	) -> list[models.Alert]:
61	"""Get all alerts for a given user."""
62	if user.id is None:	1✔
NEW 63	raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")	×
64
65	async with self.session_maker() as session:	1✔
66	query = (	1✔
67	select(schemas.AlertORM)
68	.where(schemas.AlertORM.user_id == user.id)
69	.where(schemas.AlertORM.resolved_at.is_(None))
70	)
71
72	if session_name is not None:	1✔
73	query = query.where(schemas.AlertORM.session_name == session_name)	1✔
74
75	query = query.order_by(schemas.AlertORM.id.desc())	1✔
76	alerts = await session.scalars(query)	1✔
77	alert_list = alerts.all()	1✔
78	return [alert.dump() for alert in alert_list]	1✔
79
80	async def update_alert(self, user: base_models.APIUser, alert_id: ULID, patch: models.AlertPatch) -> models.Alert:	2✔
81	"""Update an existing alert in the database."""
82	if user.id is None:	1✔
NEW 83	raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")	×
84	if not user.is_admin:	1✔
NEW 85	raise errors.ForbiddenError(message="You do not have the required permissions for this operation.")	×
86
87	async with self.session_maker() as session, session.begin():	1✔
88	res = await session.scalars(select(schemas.AlertORM).where(schemas.AlertORM.id == alert_id))	1✔
89	alert_orm = res.one_or_none()	1✔
90	if alert_orm is None:	1✔
91	raise errors.MissingResourceError(message=f"Alert with id '{alert_id}' not found.")	1✔
92
NEW 93	self.__update_alert(alert_orm, patch)	×
NEW 94	return alert_orm.dump()	×
95
96	def __update_alert(self, alert: schemas.AlertORM, update: models.AlertPatch) -> None:	2✔
NEW 97	if update.resolved is True:	×
NEW 98	alert.resolved_at = datetime.now(UTC)	×
99
100	async def get_alerts_by_properties(	2✔
101	self,
102	user: base_models.APIUser,
103	alert_id: ULID \| None,
104	user_id: str \| None,
105	title: str \| None,
106	message: str \| None,
107	session_name: str \| None,
108	created_at: datetime \| None,
109	resolved_at: datetime \| None,
110	) -> list[models.Alert]:
111	"""Get alerts by their properties."""
NEW 112	if user.id is None:	×
NEW 113	raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.")	×
114
NEW 115	async with self.session_maker() as session:	×
NEW 116	query = select(schemas.AlertORM)	×
117
NEW 118	if alert_id is not None:	×
NEW 119	query = query.where(schemas.AlertORM.id == alert_id)	×
120
NEW 121	if created_at is not None:	×
NEW 122	query = query.where(schemas.AlertORM.creation_date == created_at)	×
123
NEW 124	if title is None:	×
NEW 125	query = query.where(schemas.AlertORM.title.is_(None))	×
126	else:
NEW 127	query = query.where(schemas.AlertORM.title == title)	×
128
NEW 129	if message is None:	×
NEW 130	query = query.where(schemas.AlertORM.message.is_(None))	×
131	else:
NEW 132	query = query.where(schemas.AlertORM.message == message)	×
133
NEW 134	if resolved_at is None:	×
NEW 135	query = query.where(schemas.AlertORM.resolved_at.is_(None))	×
136	else:
NEW 137	query = query.where(schemas.AlertORM.resolved_at == resolved_at)	×
138
NEW 139	if session_name is None:	×
NEW 140	query = query.where(schemas.AlertORM.session_name.is_(None))	×
141	else:
NEW 142	query = query.where(schemas.AlertORM.session_name == session_name)	×
143
NEW 144	if user_id is None:	×
NEW 145	query = query.where(schemas.AlertORM.user_id.is_(None))	×
146	else:
NEW 147	query = query.where(schemas.AlertORM.user_id == user_id)	×
148
NEW 149	res = await session.scalars(query)	×
NEW 150	alert_list = res.all()	×
NEW 151	return [alert.dump() for alert in alert_list]	×

SwissDataScienceCenter / renku-data-services / 19325323565

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous