python-tools/async_api_client.py at main · sha2fiddy/python-tools · 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
from time import time
from typing import Type
from asyncio import TaskGroup, run  # TaskGroup object was added in Python 3.11

from httpx import Client, AsyncClient, HTTPError

"""
# nest_asyncio is needed for async code in some environments like Jupyter
# nest_asyncio requires full asyncio package, your mileage may vary
import asyncio
from nest_asyncio import apply as nest
nest()
"""

# TODO: implement asyncio ExceptionGroup

class API:
    """
    A base client for making API calls using httpx
    and modern features of the asyncio library.
    Includes a call method for a single synchronous API call,
    an async_calls method for multiple asyncronous API calls,
    and an async_calls_wrapper method to use async_calls synchronously.
    """
    def __init__(self, url:str, http2:bool=False):
        """
        Requires API base url to initialize.

        param url: string, base url of API
        param http2: bool, default False, improves async calls if server supports http2
        """
        self.url = url
        self.http2 = http2


    def call(self, endpoint:str='', params:dict=None, html:bool=False) -> str:
        """
        Method for making a single, synchronous API call.
        Takes an endpoint string and returns response text.

        param endpoint: string, API endpoint to follow base url
        param params: dict, default None, optional params object to pass to the API
        param html: bool, default False, indicicates if HTML docstring responses are accepted
        """
        try:
            with Client(http2=self.http2) as client:
                r = client.get(self.url + endpoint, params=params)
                r.raise_for_status()
                if not html and r.text[:15] == '<!doctype html>':
                    print('Unexpetedly received HTML response: ' + self.url + endpoint)
                    print(r.text[:10000])
                    return None
                elif html and r.text[:15] != '<!doctype html>':
                    print('Expected HTML response but receieved:')
                    print(r.text[:10000])
                    return None
                else:
                    return r.text

        except HTTPError as e:
            raise SystemError(e)


    async def _async_call(self, client:Type[AsyncClient], endpoint:str,
                          params:dict=None, html:bool=False) -> str:
        """
        Hidden method for making a single API call asynchronously.
        Meant to be batched with the async_calls method.
        Takes an endpoint string and returns response text.

        param client, httpx AsyncClient object
        param endpoint: string, API endpoint to follow base url
        param params: dict, default None, optional params object to pass to the API
        param html: bool, default False, indicicates if HTML docstring responses are accepted
        """
        try:
            r = await client.get(self.url + endpoint, params=params)
            r.raise_for_status()
            if not html and r.text[:15] == '<!doctype html>':
                print('Unexpetedly received HTML response: ' + self.url + endpoint)
                print(r.text[:10000] + '\n' + '...')
                return None
            else:
                return r.text

        except HTTPError as e:
            raise SystemError(e)


    async def async_calls(self, endpoints:list, params:dict=None, html:bool=False) -> list:
        """
        Method for sending multiple API calls asynchronously.
        Takes a list of endpoint strings and returns a list of response texts.
        Requires the httpx AsyncClient object as the API client engine.
        The asyncio TaskGroup object was introduced in Python 3.11.

        param: endpoints: list, list of endpoint strings to follow base url
        param params: dict, default None, optional params object to pass to the API
        param html: bool, default False, indicicates if HTML docstring responses are accepted
        """
        try:
            print(f'Making {len(endpoints)} calls to: ' + self.url + ' ...')
            start_time = time()
            async with AsyncClient(http2=self.http2) as client:
                async with TaskGroup() as tg:
                    tasks = [tg.create_task(self._async_call(client, e, params, html)) for e in endpoints]
                print('Completed in: %s seconds' % (time() - start_time) + '\n')
                return [t.result() for t in tasks]

        except:
            raise Exception('Error communicating with API')


    def async_calls_wrapper(self, endpoints:list, params:dict=None, html:bool=False) -> list:
        """
        Wrapper method for executing the async_calls
        method without importing asyncio.run.
        Takes a list of endpoint strings and returns a list of response texts.

        param: endpoints: list, list of endpoint strings to follow base url
        param params: dict, default None, optional params object to pass to the API
        param html: bool, default False, indicicates if HTML docstring responses are accepted
        """
        try:
            return run(self.async_calls(endpoints, params, html))

        except:
            raise Exception('Error executing async_calls method')


"""
# Example usage:
url = 'https://mempool.space/api'
api = API(url)

blockheight = 777777
endpoint = f'/block-height/{blockheight}'
blockhash = api.call(endpoint)

endpoints = [
    f'/block/{blockhash}',
    f'/block/{blockhash}/status',
    f'/block/{blockhash}/txids'
]
responses = api.async_calls_wrapper(endpoints)

# OR
from asyncio import run
responses = run(api.async_calls(endpoints))
"""