mpt-api-python-client/mpt_api_client/http/mixins/collection_mixin.py at 92149de3503dd65c15c49828efa2feef7a7e36b0 · softwareone-platform/mpt-api-python-client · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
from collections.abc import AsyncIterator, Iterator

from mpt_api_client.http.mixins.queryable_mixin import QueryableMixin
from mpt_api_client.http.types import Response
from mpt_api_client.models import Collection
from mpt_api_client.models import Model as BaseModel


class CollectionMixin[Model: BaseModel](QueryableMixin):
    """Mixin providing collection functionality."""

    def fetch_page(self, limit: int = 100, offset: int = 0) -> Collection[Model]:
        """Fetch one page of resources.

        Returns:
            Collection of resources.
        """
        response = self._fetch_page_as_response(limit=limit, offset=offset)
        return self.make_collection(response)  # type: ignore[attr-defined, no-any-return]

    def fetch_one(self) -> Model:
        """Fetch one resource, expect exactly one result.

        Returns:
            One resource.

        Raises:
            ValueError: If the total matching records are not exactly one.
        """
        response = self._fetch_page_as_response(limit=1, offset=0)
        resource_list = self.make_collection(response)  # type: ignore[attr-defined]
        total_records = len(resource_list)
        if resource_list.meta:
            total_records = resource_list.meta.pagination.total
        if total_records == 0:
            raise ValueError("Expected one result, but got zero results")
        if total_records > 1:
            raise ValueError(f"Expected one result, but got {total_records} results")

        return resource_list[0]  # type: ignore[no-any-return]

    def iterate(self, batch_size: int = 100) -> Iterator[Model]:
        """Iterate over all resources, yielding GenericResource objects.

        Args:
            batch_size: Number of resources to fetch per request

        Returns:
            Iterator of resources.
        """
        offset = 0
        limit = batch_size  # Default page size

        while True:
            response = self._fetch_page_as_response(limit=limit, offset=offset)
            items_collection = self.make_collection(response)  # type: ignore[attr-defined]
            yield from items_collection

            if not items_collection.meta:
                break
            if not items_collection.meta.pagination.has_next():
                break
            offset = items_collection.meta.pagination.next_offset()

    def _fetch_page_as_response(self, limit: int = 100, offset: int = 0) -> Response:
        """Fetch one page of resources.

        Returns:
            Response object.

        Raises:
            HTTPStatusError: if the response status code is not 200.
        """
        pagination_params: dict[str, int] = {"limit": limit, "offset": offset}
        return self.http_client.request("get", self.build_path(pagination_params))  # type: ignore[attr-defined, no-any-return]


class AsyncCollectionMixin[Model: BaseModel](QueryableMixin):
    """Async mixin providing collection functionality."""

    async def fetch_page(self, limit: int = 100, offset: int = 0) -> Collection[Model]:
        """Fetch one page of resources.

        Returns:
            Collection of resources.
        """
        response = await self._fetch_page_as_response(limit=limit, offset=offset)
        return self.make_collection(response)  # type: ignore[no-any-return,attr-defined]

    async def fetch_one(self) -> Model:
        """Fetch one resource, expect exactly one result.

        Returns:
            One resource.

        Raises:
            ValueError: If the total matching records are not exactly one.
        """
        response = await self._fetch_page_as_response(limit=1, offset=0)
        resource_list = self.make_collection(response)  # type: ignore[attr-defined]
        total_records = len(resource_list)
        if resource_list.meta:
            total_records = resource_list.meta.pagination.total
        if total_records == 0:
            raise ValueError("Expected one result, but got zero results")
        if total_records > 1:
            raise ValueError(f"Expected one result, but got {total_records} results")

        return resource_list[0]  # type: ignore[no-any-return]

    async def iterate(self, batch_size: int = 100) -> AsyncIterator[Model]:
        """Iterate over all resources, yielding GenericResource objects.

        Args:
            batch_size: Number of resources to fetch per request

        Returns:
            Iterator of resources.
        """
        offset = 0
        limit = batch_size  # Default page size

        while True:
            response = await self._fetch_page_as_response(limit=limit, offset=offset)
            items_collection = self.make_collection(response)  # type: ignore[attr-defined]
            for resource in items_collection:
                yield resource

            if not items_collection.meta:
                break
            if not items_collection.meta.pagination.has_next():
                break
            offset = items_collection.meta.pagination.next_offset()

    async def _fetch_page_as_response(self, limit: int = 100, offset: int = 0) -> Response:
        """Fetch one page of resources.

        Returns:
            Response object.

        Raises:
            HTTPStatusError: if the response status code is not 200.
        """
        pagination_params: dict[str, int] = {"limit": limit, "offset": offset}
        return await self.http_client.request("get", self.build_path(pagination_params))  # type: ignore[attr-defined,no-any-return]