asyncapis/asyncapis.py at master · d4tech/asyncapis · 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import asyncio
import typing
from asyncio import AbstractEventLoop

import httpx
import json
import asyncpool
import dataclasses

from httpx.content_streams import RequestData, RequestFiles

from dfsrereinstate.utils import logger
from httpx.auth import AuthTypes
from httpx.config import TimeoutTypes, UnsetType, UNSET
from httpx.models import URLTypes, QueryParamTypes, HeaderTypes, CookieTypes
from linkedin.http import tls
from dfsrereinstate.utils import load_config


@dataclasses.dataclass(eq=False)
class AsyncCall:
    get_results: dict = dataclasses.field(init=False)
    post_results: dict = dataclasses.field(init=False)
    get_inputs: list = dataclasses.field(init=False)
    post_inputs: list = dataclasses.field(init=False)

    def __post_init__(self):
        self.get_results = {}
        self.post_results = {}
        self.get_inputs = []
        self.post_inputs = []
        self.configs = load_config()
        self.log = logger.getLoggerInstance(__name__, self.configs.LOG_LEVEL)

    async def _async_get_api_call(self, i, **kwargs,):
        url = kwargs.get('url')
        retry = kwargs.pop('retry')
        # configs = load_config()
        # log = logger.getLoggerInstance(__name__, configs.LOG_LEVEL)
        async with httpx.AsyncClient(verify=tls.ca_bundle_path()) as client:
            try:
                response = await client.get(**kwargs)
                if response.status_code == 200:
                    try:
                        data = response.json()
                        self.get_results[i] = data
                    except json.decoder.JSONDecodeError:
                        data = response.text
                        self.get_results[i] = data
                        self.log.exception('Error occurred while converting to json object')
                else:
                    self.log.error(f'Status_code: {response.status_code}, response: {response.text}')
            except httpx.exceptions.HTTPError as e:
                if (retry - 1) > 0:
                    kwargs['retry'] = retry - 1
                    await self._async_get_api_call(i, **kwargs)
                else:
                    self.get_results[i] = None
                    self.log.info(f'Error occurred while making api call. Url: {url}. exception message: {e}')

    async def _async_post_api_call(self, i, **kwargs,):
        url = kwargs.get('url')
        retry = kwargs.pop('retry')
        # configs = load_config()
        # log = logger.getLoggerInstance(__name__, configs.LOG_LEVEL)
        async with httpx.AsyncClient(verify=tls.ca_bundle_path()) as client:
            try:
                response = await client.post(**kwargs)
                if response.status_code == 200:
                    try:
                        data = response.json()
                        self.post_results[i] = data
                    except json.decoder.JSONDecodeError:
                        data = response.text
                        self.post_results[i] = data
                        self.log.exception('Error occurred while converting to json object')
                else:
                    self.log.error(f'Status_code: {response.status_code}, response: {response.text}')
            except httpx.exceptions.HTTPError as e:
                if (retry - 1) > 0:
                    kwargs['retry'] = retry - 1
                    await self._async_post_api_call(i, **kwargs)
                else:
                    self.post_results[i] = None
                    print(e)
                    # self.log.exception(f'Error occurred while making api call. Url: {url}. exception message: {e}')

    async def _make_get_api_call(self,
                                 urls_params: list,
                                 loop: AbstractEventLoop,
                                 num_workers: int = 10,
                                 ):
        async with asyncpool.AsyncPool(loop, num_workers=num_workers, name="dfsre-reinstate-pool", logger=self.log,
                                       worker_co=self._async_get_api_call) as pool:
            for i, urls_param in enumerate(urls_params):
                await pool.push(i, **urls_param)
        return self.get_results

    async def _make_post_api_call(self,
                                  urls_params: list,
                                  loop: AbstractEventLoop,
                                  num_workers: int = 10,
                                  ):
        async with asyncpool.AsyncPool(loop, num_workers=num_workers, name="dfsre-reinstate-pool", logger=self.log,
                                       worker_co=self._async_post_api_call) as pool:
            for i, urls_param in enumerate(urls_params):
                await pool.push(i, **urls_param)
        return self.post_results

    def get(self):
        loop = asyncio.new_event_loop()
        try:
            result = loop.run_until_complete(self._make_get_api_call(urls_params=self.get_inputs, loop=loop))
        finally:
            loop.close()
        self.get_results = {}  # resetting the result value
        self.get_inputs = []  # resetting the result value
        return result

    def post(self):
        loop = asyncio.new_event_loop()
        try:
            result = loop.run_until_complete(self._make_post_api_call(urls_params=self.post_inputs, loop=loop))
        finally:
            loop.close()
        self.post_results = {}  # resetting the result value
        self.post_inputs = []  # resetting the result value
        return result

    def push_get_http_param(self, url: URLTypes,
                            params: QueryParamTypes = None,
                            headers: HeaderTypes = None,
                            cookies: CookieTypes = None,
                            auth: AuthTypes = None,
                            allow_redirects: bool = True,
                            retry: int = 1,
                            timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
                            ):
        self.get_inputs.append({
            'url': url,
            'params': params,
            'headers': headers,
            'cookies': cookies,
            'auth': auth,
            'allow_redirects': allow_redirects,
            'retry': retry,
            'timeout': timeout,
        })

    def push_post_http_param(self, url: URLTypes,
                             data: RequestData = None,
                             files: RequestFiles = None,
                             params: QueryParamTypes = None,
                             headers: HeaderTypes = None,
                             cookies: CookieTypes = None,
                             auth: AuthTypes = None,
                             allow_redirects: bool = True,
                             retry: int = 1,
                             timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
                             ):
        self.post_inputs.append({
            'url': url,
            'data': data,
            'files': files,
            'params': params,
            'headers': headers,
            'cookies': cookies,
            'auth': auth,
            'allow_redirects': allow_redirects,
            'retry': retry,
            'timeout': timeout,
        })