Skip to content

Commit

Permalink
Add pagination types
Browse files Browse the repository at this point in the history
Signed-off-by: Flora <flora.hofmann@frequenz.com>
  • Loading branch information
flora-hofmann-frequenz committed Feb 8, 2024
1 parent 1a034c0 commit 83ff1f1
Show file tree
Hide file tree
Showing 2 changed files with 125 additions and 0 deletions.
25 changes: 25 additions & 0 deletions src/frequenz/client/common/pagination/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
# License: MIT
# Copyright © 2023 Frequenz Energy-as-a-Service GmbH

"""Common code and utilities for Frequenz API clients.
TODO(cookiecutter): Add a more descriptive module description.
"""


# TODO(cookiecutter): Remove this function
def delete_me(*, blow_up: bool = False) -> bool:
"""Do stuff for demonstration purposes.
Args:
blow_up: If True, raise an exception.
Returns:
True if no exception was raised.
Raises:
RuntimeError: if blow_up is True.
"""
if blow_up:
raise RuntimeError("This function should be removed!")
return True
100 changes: 100 additions & 0 deletions src/frequenz/client/common/pagination/pagination.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
# License: MIT
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH

"""Module to define the pagination used with the common client."""

from __future__ import annotations # required for constructor type hinting

import logging
from dataclasses import dataclass

# pylint: disable=import-error, no-name-in-module
from frequenz.api.common.v1.pagination_info_pb2 import (
PaginationInfo as PBPaginationInfo,
)

# pylint: disable=import-error, no-name-in-module
from frequenz.api.common.v1.pagination_params_pb2 import (
PaginationParams as PBPaginationParams,
)

# Set up logging
_logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class PaginationParams:
"""
Parameters for paginating list requests.
Args:
page_size: The maximum number of results to be returned per request.
page_token: The token identifying a specific page of the list results.
"""

page_size: int | None = None
page_token: str | None = None

@classmethod
def from_pb(cls, pagination_params: PBPaginationParams) -> PaginationParams:
"""Convert a protobuf PaginationParams to PaginationParams object.
Args:
pagination_params: PaginationParams to convert.
Returns:
PaginationParams object corresponding to the protobuf message.
"""
return cls(
page_size=pagination_params.page_size,
page_token=pagination_params.page_token,
)

def to_pb(self) -> PBPaginationParams:
"""Convert a PaginationParams object to protobuf PaginationParams.
Returns:
Protobuf message corresponding to the PaginationParams object.
"""
return PBPaginationParams(
page_size=self.page_size,
page_token=self.page_token,
)


@dataclass(frozen=True)
class PaginationInfo:
"""
Information about the pagination of a list request.
Args:
total_items: The total number of items that match the request.
next_page_token: The token identifying the next page of results.
"""

total_items: int
next_page_token: str | None = None

@classmethod
def from_pb(cls, pagination_info: PBPaginationInfo) -> PaginationInfo:
"""Convert a protobuf PaginationInfo to PaginationInfo object.
Args:
pagination_info: PaginationInfo to convert.
Returns:
PaginationInfo object corresponding to the protobuf message.
"""
return cls(
total_items=pagination_info.total_items,
next_page_token=pagination_info.next_page_token,
)

def to_pb(self) -> PBPaginationInfo:
"""Convert a PaginationInfo object to protobuf PaginationInfo.
Returns:
Protobuf message corresponding to the PaginationInfo object.
"""
return PBPaginationInfo(
total_items=self.total_items,
next_page_token=self.next_page_token,
)

0 comments on commit 83ff1f1

Please sign in to comment.