11508228886

Committed 24 Oct 2024 10:09PM UTC coverage: 99.859% (-0.1%) from 100.0%

Build # 11508228886

Build Type

Pull #33

github

Committed by

ape364

Commit Message

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

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

Run Details

41 of 42 new or added lines in 5 files covered. (97.62%)

1 existing line in 1 file now uncovered.

708 of 709 relevant lines covered (99.86%)

3.99 hits per line

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

98.28

/aioetherscan/modules/extra/generators/generator_utils.py

import inspect
import sys
from itertools import count
from typing import Callable, Any, Optional, TYPE_CHECKING, AsyncIterator

from aioetherscan.exceptions import EtherscanClientApiError
from aioetherscan.modules.extra.generators.blocks_parser import BlocksParser, Transfer

if TYPE_CHECKING:  # pragma: no cover
    from aioetherscan import Client


class GeneratorUtils:
    _DEFAULT_START_BLOCK: int = 0
    _DEFAULT_END_BLOCK: int = sys.maxsize
    _DEFAULT_BLOCKS_LIMIT: int = 2048
    _DEFAULT_BLOCKS_LIMIT_DIVIDER: int = 2

    def __init__(self, client: 'Client') -> None:
        self._client = client

    async def token_transfers(
        self,
        contract_address: str = None,
        address: str = None,
        start_block: int = _DEFAULT_START_BLOCK,
        end_block: int = _DEFAULT_END_BLOCK,
        blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
        blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
    ) -> AsyncIterator[Transfer]:
        parser_params = self._get_parser_params(self._client.account.token_transfers, locals())
        async for transfer in self._parse_by_blocks(**parser_params):
            yield transfer

    async def normal_txs(
        self,
        address: str,
        start_block: int = _DEFAULT_START_BLOCK,
        end_block: int = _DEFAULT_END_BLOCK,
        blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
        blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
    ) -> AsyncIterator[Transfer]:
        parser_params = self._get_parser_params(self._client.account.normal_txs, locals())
        async for transfer in self._parse_by_blocks(**parser_params):
            yield transfer

    async def internal_txs(
        self,
        address: str,
        start_block: int = _DEFAULT_START_BLOCK,
        end_block: int = _DEFAULT_END_BLOCK,
        blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
        blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
        txhash: Optional[str] = None,
    ) -> AsyncIterator[Transfer]:
        parser_params = self._get_parser_params(self._client.account.internal_txs, locals())
        async for transfer in self._parse_by_blocks(**parser_params):
            yield transfer

    async def mined_blocks(
        self, address: str, blocktype: str, offset: int = 10_000
    ) -> AsyncIterator[Transfer]:
        parser_params = self._get_parser_params(self._client.account.mined_blocks, locals())
        async for transfer in self._parse_by_pages(**parser_params):
            yield transfer

    async def _parse_by_blocks(
        self,
        api_method: Callable,
        request_params: dict[str, Any],
        start_block: int,
        end_block: int,
        blocks_limit: int,
        blocks_limit_divider: int,
    ) -> AsyncIterator[Transfer]:
        blocks_parser = self._get_blocks_parser(
            api_method, request_params, start_block, end_block, blocks_limit, blocks_limit_divider
        )
        async for tx in blocks_parser.txs_generator():
            yield tx

    @staticmethod
    async def _parse_by_pages(
        api_method: Callable, request_params: dict[str, Any]
    ) -> AsyncIterator[Transfer]:
        page = count(1)
        while True:
            request_params['page'] = next(page)
            try:
                result = await api_method(**request_params)
            except EtherscanClientApiError as e:
                if e.message == 'No transactions found':
                    return
                raise
            else:
                for row in result:
                    yield row

    @staticmethod
    def _without_keys(params: dict, excluded_keys: tuple[str, ...] = ('self',)) -> dict:
        return {k: v for k, v in params.items() if k not in excluded_keys}

    def _get_parser_params(self, api_method: Callable, params: dict[str, Any]) -> dict[str, Any]:
        request_params = self._get_request_params(api_method, params)
        return self._without_keys(
            dict(
                api_method=api_method,
                request_params=request_params,
                **{k: v for k, v in params.items() if k not in request_params},
            )
        )

    def _get_request_params(self, api_method: Callable, params: dict[str, Any]) -> dict[str, Any]:
        api_method_params = inspect.getfullargspec(api_method).args
        return self._without_keys(
            {k: v for k, v in params.items() if k in api_method_params},
            ('self', 'start_block', 'end_block'),
        )

    @staticmethod
    def _get_blocks_parser(
        api_method: Callable,
        request_params: dict[str, Any],
        start_block: int,
        end_block: int,
        blocks_limit: int,
        blocks_limit_divider: int,
    ) -> BlocksParser:
        return BlocksParser(
            api_method, request_params, start_block, end_block, blocks_limit, blocks_limit_divider
        )

1	import inspect	4✔
2	import sys	4✔
3	from itertools import count	4✔
4	from typing import Callable, Any, Optional, TYPE_CHECKING, AsyncIterator	4✔
5
6	from aioetherscan.exceptions import EtherscanClientApiError	4✔
7	from aioetherscan.modules.extra.generators.blocks_parser import BlocksParser, Transfer	4✔
8
9	if TYPE_CHECKING: # pragma: no cover
10	from aioetherscan import Client
11
12
13	class GeneratorUtils:	4✔
14	_DEFAULT_START_BLOCK: int = 0	4✔
15	_DEFAULT_END_BLOCK: int = sys.maxsize	4✔
16	_DEFAULT_BLOCKS_LIMIT: int = 2048	4✔
17	_DEFAULT_BLOCKS_LIMIT_DIVIDER: int = 2	4✔
18
19	def __init__(self, client: 'Client') -> None:	4✔
20	self._client = client	4✔
21
22	async def token_transfers(	4✔
23	self,
24	contract_address: str = None,
25	address: str = None,
26	start_block: int = _DEFAULT_START_BLOCK,
27	end_block: int = _DEFAULT_END_BLOCK,
28	blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
29	blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
30	) -> AsyncIterator[Transfer]:
31	parser_params = self._get_parser_params(self._client.account.token_transfers, locals())	4✔
32	async for transfer in self._parse_by_blocks(**parser_params):	4✔
33	yield transfer	4✔
34
35	async def normal_txs(	4✔
36	self,
37	address: str,
38	start_block: int = _DEFAULT_START_BLOCK,
39	end_block: int = _DEFAULT_END_BLOCK,
40	blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
41	blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
42	) -> AsyncIterator[Transfer]:
43	parser_params = self._get_parser_params(self._client.account.normal_txs, locals())	4✔
44	async for transfer in self._parse_by_blocks(**parser_params):	4✔
45	yield transfer	4✔
46
47	async def internal_txs(	4✔
48	self,
49	address: str,
50	start_block: int = _DEFAULT_START_BLOCK,
51	end_block: int = _DEFAULT_END_BLOCK,
52	blocks_limit: int = _DEFAULT_BLOCKS_LIMIT,
53	blocks_limit_divider: int = _DEFAULT_BLOCKS_LIMIT_DIVIDER,
54	txhash: Optional[str] = None,
55	) -> AsyncIterator[Transfer]:
56	parser_params = self._get_parser_params(self._client.account.internal_txs, locals())	4✔
57	async for transfer in self._parse_by_blocks(**parser_params):	4✔
58	yield transfer	4✔
59
60	async def mined_blocks(	4✔
61	self, address: str, blocktype: str, offset: int = 10_000
62	) -> AsyncIterator[Transfer]:
63	parser_params = self._get_parser_params(self._client.account.mined_blocks, locals())	4✔
64	async for transfer in self._parse_by_pages(**parser_params):	4✔
65	yield transfer	4✔
66
67	async def _parse_by_blocks(	4✔
68	self,
69	api_method: Callable,
70	request_params: dict[str, Any],
71	start_block: int,
72	end_block: int,
73	blocks_limit: int,
74	blocks_limit_divider: int,
75	) -> AsyncIterator[Transfer]:
76	blocks_parser = self._get_blocks_parser(	4✔
77	api_method, request_params, start_block, end_block, blocks_limit, blocks_limit_divider
78	)
79	async for tx in blocks_parser.txs_generator():	4✔
80	yield tx	4✔
81
82	@staticmethod	4✔
83	async def _parse_by_pages(	4✔
84	api_method: Callable, request_params: dict[str, Any]
85	) -> AsyncIterator[Transfer]:
86	page = count(1)	4✔
NEW UNCOV 87	while True:	×
88	request_params['page'] = next(page)	4✔
89	try:	4✔
90	result = await api_method(**request_params)	4✔
91	except EtherscanClientApiError as e:	4✔
92	if e.message == 'No transactions found':	4✔
93	return	4✔
94	raise	4✔
95	else:
96	for row in result:	4✔
97	yield row	4✔
98
99	@staticmethod	4✔
100	def _without_keys(params: dict, excluded_keys: tuple[str, ...] = ('self',)) -> dict:	4✔
101	return {k: v for k, v in params.items() if k not in excluded_keys}	4✔
102
103	def _get_parser_params(self, api_method: Callable, params: dict[str, Any]) -> dict[str, Any]:	4✔
104	request_params = self._get_request_params(api_method, params)	4✔
105	return self._without_keys(	4✔
106	dict(
107	api_method=api_method,
108	request_params=request_params,
109	**{k: v for k, v in params.items() if k not in request_params},
110	)
111	)
112
113	def _get_request_params(self, api_method: Callable, params: dict[str, Any]) -> dict[str, Any]:	4✔
114	api_method_params = inspect.getfullargspec(api_method).args	4✔
115	return self._without_keys(	4✔
116	{k: v for k, v in params.items() if k in api_method_params},
117	('self', 'start_block', 'end_block'),
118	)
119
120	@staticmethod	4✔
121	def _get_blocks_parser(	4✔
122	api_method: Callable,
123	request_params: dict[str, Any],
124	start_block: int,
125	end_block: int,
126	blocks_limit: int,
127	blocks_limit_divider: int,
128	) -> BlocksParser:
129	return BlocksParser(	4✔
130	api_method, request_params, start_block, end_block, blocks_limit, blocks_limit_divider
131	)

ape364 / aioetherscan / 11508228886

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