Source code for codegrade._api.notification

"""The endpoints for notification objects.

SPDX-License-Identifier: AGPL-3.0-only OR BSD-3-Clause-Clear
"""

import os
import typing as t

import cg_request_args as rqa
from cg_maybe import Maybe, Nothing

from .. import parsers, utils

if t.TYPE_CHECKING or os.getenv("CG_EAGERIMPORT", False):
    from .. import client
    from ..models.has_unread_notifcation_json import HasUnreadNotifcationJSON
    from ..models.notification import Notification
    from ..models.notifications_json import NotificationsJSON
    from ..models.patch_all_notification_data import PatchAllNotificationData
    from ..models.patch_notification_data import PatchNotificationData


_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")


[docs]class NotificationService(t.Generic[_ClientT]): __slots__ = ("__client",) def __init__(self, client: _ClientT) -> None: self.__client = client
[docs] def get_all( self: "NotificationService[client.AuthenticatedClient]", *, has_unread: "bool" = False, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Union[NotificationsJSON, HasUnreadNotifcationJSON]": """Get all notifications for the current user. :param has_unread: If considered true a short digest will be sent, i.e. a single object with one key `has_unread` with a boolean value. Please use this if you simply want to check if there are unread notifications. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: Either a list of notifications or a `HasUnreadNotifcationJSON` based on the `has_unread` query parameter. """ url = "/api/v1/notifications/" params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "has_unread": utils.to_dict(has_unread), } with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.has_unread_notifcation_json import ( HasUnreadNotifcationJSON, ) from ..models.notifications_json import NotificationsJSON return parsers.JsonResponseParser( parsers.make_union( parsers.ParserFor.make(NotificationsJSON), parsers.ParserFor.make(HasUnreadNotifcationJSON), ) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def patch_all( self: "NotificationService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchAllNotificationData"], *, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "NotificationsJSON": """Update the read status of multiple notifications. :param json_body: The body of the request. See :class:`.PatchAllNotificationData` for information about the possible fields. You can provide this data as a :class:`.PatchAllNotificationData` or as a dictionary. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The updated notifications in the same order as given in the body. """ url = "/api/v1/notifications/" params = extra_parameters or {} with self.__client as client: resp = client.http.patch( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.notifications_json import NotificationsJSON return parsers.JsonResponseParser( parsers.ParserFor.make(NotificationsJSON) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def patch( self: "NotificationService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchNotificationData"], *, notification_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "Notification": """Update the read status for the given notification. :param json_body: The body of the request. See :class:`.PatchNotificationData` for information about the possible fields. You can provide this data as a :class:`.PatchNotificationData` or as a dictionary. :param notification_id: The id of the notification to update. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The updated notification. """ url = "/api/v1/notifications/{notificationId}".format( notificationId=notification_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.patch( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.notification import NotificationParser return parsers.JsonResponseParser(NotificationParser).try_parse( resp ) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )