Skip to content

Notification module

Queries

Set of Notification queries.

Source code in kili/queries/notification/__init__.py
class QueriesNotification:
    """Set of Notification queries."""

    # pylint: disable=too-many-arguments,too-many-locals

    def __init__(self, auth):
        """Initialize the subclass.

        Args:
            auth: KiliAuth object
        """
        self.auth = auth

    # pylint: disable=dangerous-default-value
    @Compatible(['v2'])
    @typechecked
    def notifications(self,
                      fields: List[str] = ['createdAt', 'hasBeenSeen',
                                           'id', 'message', 'status', 'userID'],
                      first: Optional[int] = None,
                      has_been_seen: Optional[bool] = None,
                      notification_id: Optional[str] = None,
                      skip: int = 0,
                      user_id: Optional[str] = None,
                      disable_tqdm: bool = False,
                      as_generator: bool = False) -> Union[List[dict], Generator[dict, None, None]]:
        # pylint: disable=line-too-long
        """Get a generator or a list of notifications respecting a set of criteria.

        Args:
            fields: All the fields to request among the possible fields for the notifications
                See [the documentation](https://docs.kili-technology.com/reference/graphql-api#notification) for all possible fields.
            first: Number of notifications to query
            has_been_seen: If the notifications returned should have been seen.
            notification_id: If given, will return the notification which has this id
            skip: Number of notifications to skip (they are ordered by their date of creation,
                first to last).
            user_id: If given, returns the notifications of a specific user
            disable_tqdm: If `True`, the progress bar will be disabled
            as_generator: If `True`, a generator on the notifications is returned.

        Returns:
            A result object which contains the query if it was successful,
                or an error message.
        """

        count_args = {"has_been_seen": has_been_seen, "user_id": user_id}
        disable_tqdm = disable_tqdm or as_generator or notification_id is not None
        payload_query = {
            'where': {
                'id': notification_id,
                'user': {
                    'id': user_id,
                },
                'hasBeenSeen': has_been_seen,
            },
        }
        notifications_generator = row_generator_from_paginated_calls(
            skip,
            first,
            self.count_notifications,
            count_args,
            self._query_notifications,
            payload_query,
            fields,
            disable_tqdm
        )

        if as_generator:
            return notifications_generator
        return list(notifications_generator)

    def _query_notifications(self,
                             skip: int,
                             first: int,
                             payload: dict,
                             fields: List[str]):

        payload.update({'skip': skip, 'first': first})
        _gql_notifications = gql_notifications(
            fragment_builder(fields, Notification))
        result = self.auth.client.execute(_gql_notifications, payload)
        return format_result('data', result)

    @Compatible(['v2'])
    @typechecked
    def count_notifications(self,
                            has_been_seen: Optional[bool] = None,
                            user_id: Optional[str] = None) -> int:
        """Count the number of notifications.

        Args:
            has_been_seen: Filter on notifications that have been seen.
            user_id: Filter on the notifications of a specific user

        Returns:
            The number of notifications with the parameters provided
        """
        variables = {
            'where': {
                'user': {
                    'id': user_id,
                },
                'hasBeenSeen': has_been_seen,
            },
        }
        result = self.auth.client.execute(GQL_NOTIFICATIONS_COUNT, variables)
        count = format_result('data', result)
        return count

count_notifications(self, has_been_seen=None, user_id=None)

Count the number of notifications.

Parameters:

Name Type Description Default
has_been_seen Optional[bool]

Filter on notifications that have been seen.

None
user_id Optional[str]

Filter on the notifications of a specific user

None

Returns:

Type Description
int

The number of notifications with the parameters provided

Source code in kili/queries/notification/__init__.py
@Compatible(['v2'])
@typechecked
def count_notifications(self,
                        has_been_seen: Optional[bool] = None,
                        user_id: Optional[str] = None) -> int:
    """Count the number of notifications.

    Args:
        has_been_seen: Filter on notifications that have been seen.
        user_id: Filter on the notifications of a specific user

    Returns:
        The number of notifications with the parameters provided
    """
    variables = {
        'where': {
            'user': {
                'id': user_id,
            },
            'hasBeenSeen': has_been_seen,
        },
    }
    result = self.auth.client.execute(GQL_NOTIFICATIONS_COUNT, variables)
    count = format_result('data', result)
    return count

notifications(self, fields=['createdAt', 'hasBeenSeen', 'id', 'message', 'status', 'userID'], first=None, has_been_seen=None, notification_id=None, skip=0, user_id=None, disable_tqdm=False, as_generator=False)

Get a generator or a list of notifications respecting a set of criteria.

Parameters:

Name Type Description Default
fields List[str]

All the fields to request among the possible fields for the notifications See the documentation for all possible fields.

['createdAt', 'hasBeenSeen', 'id', 'message', 'status', 'userID']
first Optional[int]

Number of notifications to query

None
has_been_seen Optional[bool]

If the notifications returned should have been seen.

None
notification_id Optional[str]

If given, will return the notification which has this id

None
skip int

Number of notifications to skip (they are ordered by their date of creation, first to last).

0
user_id Optional[str]

If given, returns the notifications of a specific user

None
disable_tqdm bool

If True, the progress bar will be disabled

False
as_generator bool

If True, a generator on the notifications is returned.

False

Returns:

Type Description
Union[List[dict], Generator[dict, NoneType]]

A result object which contains the query if it was successful, or an error message.

Source code in kili/queries/notification/__init__.py
@Compatible(['v2'])
@typechecked
def notifications(self,
                  fields: List[str] = ['createdAt', 'hasBeenSeen',
                                       'id', 'message', 'status', 'userID'],
                  first: Optional[int] = None,
                  has_been_seen: Optional[bool] = None,
                  notification_id: Optional[str] = None,
                  skip: int = 0,
                  user_id: Optional[str] = None,
                  disable_tqdm: bool = False,
                  as_generator: bool = False) -> Union[List[dict], Generator[dict, None, None]]:
    # pylint: disable=line-too-long
    """Get a generator or a list of notifications respecting a set of criteria.

    Args:
        fields: All the fields to request among the possible fields for the notifications
            See [the documentation](https://docs.kili-technology.com/reference/graphql-api#notification) for all possible fields.
        first: Number of notifications to query
        has_been_seen: If the notifications returned should have been seen.
        notification_id: If given, will return the notification which has this id
        skip: Number of notifications to skip (they are ordered by their date of creation,
            first to last).
        user_id: If given, returns the notifications of a specific user
        disable_tqdm: If `True`, the progress bar will be disabled
        as_generator: If `True`, a generator on the notifications is returned.

    Returns:
        A result object which contains the query if it was successful,
            or an error message.
    """

    count_args = {"has_been_seen": has_been_seen, "user_id": user_id}
    disable_tqdm = disable_tqdm or as_generator or notification_id is not None
    payload_query = {
        'where': {
            'id': notification_id,
            'user': {
                'id': user_id,
            },
            'hasBeenSeen': has_been_seen,
        },
    }
    notifications_generator = row_generator_from_paginated_calls(
        skip,
        first,
        self.count_notifications,
        count_args,
        self._query_notifications,
        payload_query,
        fields,
        disable_tqdm
    )

    if as_generator:
        return notifications_generator
    return list(notifications_generator)