convert_to_async.py

"""Сгенерировать асинхронную версию клиента."""

DISCLAIMER = "# ЭТО АВТОМАТИЧЕСКИ СОЗДАННАЯ КОПИЯ СИНХРОННОГО КЛИЕНТА. НЕ ИЗМЕНЯЙТЕ САМОСТОЯТЕЛЬНО #"
DISCLAIMER = f'{"#" * len(DISCLAIMER)}\n{DISCLAIMER}\n{"#" * len(DISCLAIMER)}\n\n'

REQUEST_METHODS = ('_request_wrapper', 'get', 'post', 'retrieve', 'download')


def gen_client(source: str, output: str) -> None:
    """Generate async version of client.py."""
    with open(source, 'r', encoding='UTF-8') as f:
        code = f.read()

    code = code.replace('class Client(', 'class ClientAsync(')
    code = code.replace('ExtClient(Client)', 'ExtClientAsync(ClientAsync)')
    code = code.replace('class WebClient', 'class WebClientAsync')
    code = code.replace("""from steam_trader.api import (
    Client,""", """from steam_trader.api import (
    ClientAsync,""")

    code = code.replace('def wrapper', 'async def wrapper')
    code = code.replace('result = method(', 'result = await method(')
    code = code.replace('@log\n    def', '@log\n    async def')
    code = code.replace('@property\n    def', '@property\n    async def')
    code = code.replace('self._get_request', 'await self._get_request')
    code = code.replace('self._post_request', 'await self._post_request')

    code = code.replace('def _get_request(', 'async def _get_request(')
    code = code.replace('(self._httpx_client or httpx).get(', 'await self._httpx_client.get(')
    
    code = code.replace('def _post_request(', 'async def _post_request(')
    code = code.replace('result = (self._httpx_client or httpx).post(', 'result = await self._httpx_client.post(')
    
    code = code.replace('self._httpx_client.close()', 'await self._httpx_client.aclose()')
    code = code.replace('self._httpx_client = httpx.Client(', 'self._httpx_client = httpx.AsyncClient(')
    
    code = code.replace('def __enter__(self) -> Self:', 'async def __aenter__(self) -> Self:')
    code = code.replace('def __exit__(self, exc_type, exc_val, exc_tb) -> None:', 'async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:')
    
    code = code.replace('url: str = self.base_url + method', """if not self._httpx_client:
            raise ClientError('Необходимо использовать контекст async with ClientAsync()')
        url: str = self.base_url + method""")
    
    code = code.replace("""result = (self._httpx_client or httpx).get(
            url,
            headers=headers,
            params=params,
            cookies=cookies,
            **kwargs
        )
        )""", """if not self._httpx_client:
            raise ClientError('Необходимо использовать контекст async with ClientAsync()')
        result = (self._httpx_client or httpx).get(
            url,
            headers=headers,
            params=params,
            cookies=cookies,
            **kwargs
        )""")
    code = code.replace("""result = await self._httpx_client.post(
            url,
            headers=self.headers,
            data=data
        )""", """if not self._httpx_client:
            raise ClientError('Необходимо использовать контекст async with ClientAsync()')
        result = (await self._httpx_client.post(
            url,
            headers=self.headers,
            data=data
        ))""")
    
    code = code.replace('self.get_order_book(gid)', 'await self.get_order_book(gid)')
    code = code.replace('self.get_item_info(', 'await self.get_item_info(')
    
    code = code.replace('await self.get_item_info(item.gid).filters', '(await self.get_item_info(item.gid)).filters')
    
    code = DISCLAIMER + code
    with open(output, 'w', encoding='UTF-8') as f:
        f.write(code)


if __name__ == '__main__':
    gen_client('steam_trader/api/_client.py', 'steam_trader/api/_client_async.py')
    gen_client('steam_trader/web/_client.py', 'steam_trader/web/_client_async.py')
    gen_client('steam_trader/api/ext/_client_ext.py', 'steam_trader/api/ext/_client_async_ext.py')