Skip to content

Abc

BulkRequestModel

bulk request base model

can be inherited from

:cvar _request_model: the request model that is used to make the bulk request :type _request_model: Any :cvar _requests: list of requests :type _requests: list

Source code in pyclasher/api/bulk_requests/abc.py 
 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
class BulkRequestModel:
    _request_model = ...
    _requests = None

    @property
    def request_model(self):
        return self._request_model

    @property
    def requests(self):
        return self._requests

    def __get_properties(self):
        return {
            name: prop.__get__(self)
            for name, prop in vars(self.__class__).items()
            if isinstance(prop, property)
        }

    async def request(self, client_id=None):
        self._tasks = [request.request(client_id) for request in self._requests]
        await gather(*self._tasks)
        return self

    def __len__(self):
        return len(self._requests)

    def __getitem__(self, item):
        self._requests[0].to_dict()         # test if the `to_dict()` method raises `RequestNotDone`
        if isinstance(item, int):
            return self._requests[item]
        if isinstance(item, slice):
            return (self._requests[i]
                    for i in range(*item.indices(len(self._requests))))
        raise NotImplementedError

    def __iter__(self):
        self._iter = iter(self._requests)
        return self

    def __next__(self):
        return next(self._iter)

    def __str__(self):
        return f"{self.__class__.__name__}()"

    def __repr__(self):
        props = ', '.join(
            ('='.join((key, str(value)))
             for key, value in self.__get_properties().items())
        )
        return f"{self.__class__.__name__}({props})"

request_model: Any property

property of the request model

:return: the specified request model :rtype: Any

requests: list property

property of the requests

:return: the list of the requests or None if the requests are not done yet :rtype: list

__get_properties()

private method that creates a dictionary of the properties

key: name of the property

value: value of the property

:return: a dictionary of the properties :rtype: dict

Source code in pyclasher/api/bulk_requests/abc.py 
16
17
18
19
20
21
def __get_properties(self):
    return {
        name: prop.__get__(self)
        for name, prop in vars(self.__class__).items()
        if isinstance(prop, property)
    }

request(client_id=None) async

asynchronous method that executes the requests

:return: the instance of the bulk request model :rtype: BulkRequestModel

Source code in pyclasher/api/bulk_requests/abc.py 
23
24
25
26
async def request(self, client_id=None):
    self._tasks = [request.request(client_id) for request in self._requests]
    await gather(*self._tasks)
    return self