11507868142

Committed 24 Oct 2024 09:37PM UTC coverage: 99.021% (-1.0%) from 100.0%

Build # 11507868142

Build Type

Pull #33

github

Committed by

ape364

Commit Message

feat: add api keys rotating - add network.py tests

Pull Request Pull Request #33: feat: add api keys rotating

Run Details

41 of 47 new or added lines in 4 files covered. (87.23%)

1 existing line in 1 file now uncovered.

708 of 715 relevant lines covered (99.02%)

3.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.8

/aioetherscan/network.py

import asyncio
import logging
from asyncio import AbstractEventLoop
from functools import wraps
from typing import Union, AsyncContextManager, Optional

import aiohttp
from aiohttp import ClientTimeout
from aiohttp.client import ClientSession
from aiohttp.hdrs import METH_GET, METH_POST
from aiohttp_retry import RetryOptionsBase, RetryClient
from asyncio_throttle import Throttler

from aioetherscan.exceptions import (
    EtherscanClientContentTypeError,
    EtherscanClientError,
    EtherscanClientApiError,
    EtherscanClientProxyError,
    EtherscanClientApiRateLimitError,
)
from aioetherscan.url_builder import UrlBuilder


def retry_limit_attempt(f):
    @wraps(f)
    async def inner(self, *args, **kwargs):
        attempt = 1
        max_attempts = self._url_builder.keys_count
        while True:
            try:
                return await f(self, *args, **kwargs)
            except EtherscanClientApiRateLimitError as e:
                self._logger.warning(f'Key daily limit exceeded, {attempt=}: {e}')
                attempt += 1
                if attempt > max_attempts:
                    raise e
                await asyncio.sleep(0.01)
                self._url_builder.rotate_api_key()

    return inner


class Network:
    def __init__(
        self,
        url_builder: UrlBuilder,
        loop: Optional[AbstractEventLoop],
        timeout: Optional[ClientTimeout],
        proxy: Optional[str],
        throttler: Optional[AsyncContextManager],
        retry_options: Optional[RetryOptionsBase],
    ) -> None:
        self._url_builder = url_builder

        self._loop = loop or asyncio.get_running_loop()
        self._timeout = timeout

        self._proxy = proxy

        # Defaulting to free API key rate limit
        self._throttler = throttler or Throttler(rate_limit=5, period=1.0)

        self._retry_client = None
        self._retry_options = retry_options

        self._logger = logging.getLogger(__name__)

    async def close(self):
        if self._retry_client is not None:
            await self._retry_client.close()

    @retry_limit_attempt
    async def get(self, params: dict = None) -> Union[dict, list, str]:
        return await self._request(METH_GET, params=self._url_builder.filter_and_sign(params))

    @retry_limit_attempt
    async def post(self, data: dict = None) -> Union[dict, list, str]:
        return await self._request(METH_POST, data=self._url_builder.filter_and_sign(data))

    def _get_retry_client(self) -> RetryClient:
        return RetryClient(client_session=self._get_session(), retry_options=self._retry_options)

    def _get_session(self) -> ClientSession:
        if self._timeout is not None:
            return ClientSession(loop=self._loop, timeout=self._timeout)
        return ClientSession(loop=self._loop)

    async def _request(
        self, method: str, data: dict = None, params: dict = None
    ) -> Union[dict, list, str]:
        if self._retry_client is None:
            self._retry_client = self._get_retry_client()
        session_method = getattr(self._retry_client, method.lower())

        async with self._throttler:
            async with session_method(
                self._url_builder.API_URL, params=params, data=data, proxy=self._proxy
            ) as response:
                self._logger.debug(
                    '[%s] %r %r %s', method, str(response.url), data, response.status
                )
                return await self._handle_response(response)

    async def _handle_response(self, response: aiohttp.ClientResponse) -> Union[dict, list, str]:
        try:
            response_json = await response.json()
        except aiohttp.ContentTypeError:
            raise EtherscanClientContentTypeError(response.status, await response.text())
        except Exception as e:
            raise EtherscanClientError(e)
        else:
            self._logger.debug('Response: %r', response_json)
            self._raise_if_error(response_json)
            return response_json['result']

    @staticmethod
    def _raise_if_error(response_json: dict):
        if 'status' in response_json and response_json['status'] != '1':
            message, result = response_json.get('message'), response_json.get('result')

            if 'max daily rate limit reached' in result.lower():
                raise EtherscanClientApiRateLimitError(message, result)

            raise EtherscanClientApiError(message, result)

        if 'error' in response_json:
            err = response_json['error']
            code, message = err.get('code'), err.get('message')
            raise EtherscanClientProxyError(code, message)

1	import asyncio	4✔
2	import logging	4✔
3	from asyncio import AbstractEventLoop	4✔
4	from functools import wraps	4✔
5	from typing import Union, AsyncContextManager, Optional	4✔
6
7	import aiohttp	4✔
8	from aiohttp import ClientTimeout	4✔
9	from aiohttp.client import ClientSession	4✔
10	from aiohttp.hdrs import METH_GET, METH_POST	4✔
11	from aiohttp_retry import RetryOptionsBase, RetryClient	4✔
12	from asyncio_throttle import Throttler	4✔
13
14	from aioetherscan.exceptions import (	4✔
15	EtherscanClientContentTypeError,
16	EtherscanClientError,
17	EtherscanClientApiError,
18	EtherscanClientProxyError,
19	EtherscanClientApiRateLimitError,
20	)
21	from aioetherscan.url_builder import UrlBuilder	4✔
22
23
24	def retry_limit_attempt(f):	4✔
25	@wraps(f)	4✔
26	async def inner(self, args, *kwargs):	4✔
27	attempt = 1	4✔
28	max_attempts = self._url_builder.keys_count	4✔
NEW 29	while True:	×
30	try:	4✔
31	return await f(self, args, *kwargs)	4✔
32	except EtherscanClientApiRateLimitError as e:	4✔
33	self._logger.warning(f'Key daily limit exceeded, {attempt=}: {e}')	4✔
34	attempt += 1	4✔
35	if attempt > max_attempts:	4✔
36	raise e	4✔
37	await asyncio.sleep(0.01)	4✔
38	self._url_builder.rotate_api_key()	4✔
39
40	return inner	4✔
41
42
43	class Network:	4✔
44	def __init__(	4✔
45	self,
46	url_builder: UrlBuilder,
47	loop: Optional[AbstractEventLoop],
48	timeout: Optional[ClientTimeout],
49	proxy: Optional[str],
50	throttler: Optional[AsyncContextManager],
51	retry_options: Optional[RetryOptionsBase],
52	) -> None:
53	self._url_builder = url_builder	4✔
54
55	self._loop = loop or asyncio.get_running_loop()	4✔
56	self._timeout = timeout	4✔
57
58	self._proxy = proxy	4✔
59
60	# Defaulting to free API key rate limit
61	self._throttler = throttler or Throttler(rate_limit=5, period=1.0)	4✔
62
63	self._retry_client = None	4✔
64	self._retry_options = retry_options	4✔
65
66	self._logger = logging.getLogger(__name__)	4✔
67
68	async def close(self):	4✔
69	if self._retry_client is not None:	4✔
70	await self._retry_client.close()	4✔
71
72	@retry_limit_attempt	4✔
73	async def get(self, params: dict = None) -> Union[dict, list, str]:	4✔
74	return await self._request(METH_GET, params=self._url_builder.filter_and_sign(params))	4✔
75
76	@retry_limit_attempt	4✔
77	async def post(self, data: dict = None) -> Union[dict, list, str]:	4✔
78	return await self._request(METH_POST, data=self._url_builder.filter_and_sign(data))	4✔
79
80	def _get_retry_client(self) -> RetryClient:	4✔
81	return RetryClient(client_session=self._get_session(), retry_options=self._retry_options)	4✔
82
83	def _get_session(self) -> ClientSession:	4✔
84	if self._timeout is not None:	4✔
85	return ClientSession(loop=self._loop, timeout=self._timeout)	4✔
86	return ClientSession(loop=self._loop)	4✔
87
88	async def _request(	4✔
89	self, method: str, data: dict = None, params: dict = None
90	) -> Union[dict, list, str]:
91	if self._retry_client is None:	4✔
92	self._retry_client = self._get_retry_client()	4✔
93	session_method = getattr(self._retry_client, method.lower())	4✔
94
95	async with self._throttler:	4✔
96	async with session_method(	4✔
97	self._url_builder.API_URL, params=params, data=data, proxy=self._proxy
98	) as response:
99	self._logger.debug(	4✔
100	'[%s] %r %r %s', method, str(response.url), data, response.status
101	)
102	return await self._handle_response(response)	4✔
103
104	async def _handle_response(self, response: aiohttp.ClientResponse) -> Union[dict, list, str]:	4✔
105	try:	4✔
106	response_json = await response.json()	4✔
107	except aiohttp.ContentTypeError:	4✔
108	raise EtherscanClientContentTypeError(response.status, await response.text())	4✔
109	except Exception as e:	4✔
110	raise EtherscanClientError(e)	4✔
111	else:
112	self._logger.debug('Response: %r', response_json)	4✔
113	self._raise_if_error(response_json)	4✔
114	return response_json['result']	4✔
115
116	@staticmethod	4✔
117	def _raise_if_error(response_json: dict):	4✔
118	if 'status' in response_json and response_json['status'] != '1':	4✔
119	message, result = response_json.get('message'), response_json.get('result')	4✔
120
121	if 'max daily rate limit reached' in result.lower():	4✔
122	raise EtherscanClientApiRateLimitError(message, result)	4✔
123
124	raise EtherscanClientApiError(message, result)	4✔
125
126	if 'error' in response_json:	4✔
127	err = response_json['error']	4✔
128	code, message = err.get('code'), err.get('message')	4✔
129	raise EtherscanClientProxyError(code, message)	4✔

ape364 / aioetherscan / 11507868142

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous