Writing both an async and sync client #2699
Unanswered
illeatmyhat
asked this question in
Q&A
Replies: 2 comments
-
This is the closest I've gotten. It turns out that as long as you copy most of the function signature, you're golden. from functools import wraps
from httpx import AsyncClient, Client
from typing import Callable, TypeVar, ParamSpec
import asyncio
T = TypeVar('T')
P = ParamSpec('P')
def synchronous(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def synchronized(*args: P.args, **kwargs: P.kwargs) -> T:
return asyncio.get_event_loop().run_until_complete(func(*args, **kwargs))
return synchronized
class Stuff:
client: AsyncClient
async def getter(self, url: str) -> str:
"""
foobar
:param url: hello
:return:
"""
return "hello world"
class SyncStuff(Stuff):
client: Client
@synchronous
def getter(self, url):
return super().getter(url)
if __name__ == '__main__':
print(SyncStuff().getter('foo')) |
Beta Was this translation helpful? Give feedback.
0 replies
-
OK, I figured it out. This lets me break up the functions into different files. Not much can be done about the duplication though. from httpx import AsyncClient, Client, Response
from typing import Protocol
import asyncio
class APIClient(Protocol):
client: AsyncClient | Client
async def async_get_id(self: APIClient) -> Response:
"""
hello world
:param self: ignore this parameter it'll work
"""
return await self.client.get('/')
def sync_get_id(self: APIClient) -> Response:
"""
hello world
:param self: ignore this parameter it'll work
"""
return self.client.get('/')
class AsyncAPI:
client: AsyncClient
def __init__(self, client: AsyncClient):
self.client = client
id = async_get_id
class SyncAPI:
client: Client
def __init__(self, client: Client):
self.client = client
id = sync_get_id
if __name__ == '__main__':
api = SyncAPI(Client(base_url="http://example.com/"))
# says "Parameter 'self' unfilled" at id()
# actually works anyways.
print(api.id().text)
api = AsyncAPI(AsyncClient(base_url="http://example.com"))
print(asyncio.get_event_loop().run_until_complete(api.id()).text) |
Beta Was this translation helpful? Give feedback.
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
It seems somewhat dissatisfying that a synchronous and asynchronous client is 90% the same code
I've been experimenting with
functools.partialmethod
and decorator functions to try to write a client for a very large API that can "do it all" while hiding the ugliness, so to speak, asrpc.py
's can, while maintaining type hints and docstrings, but no luck so far.What are some approaches you have taken to minimize repeating yourself?
Beta Was this translation helpful? Give feedback.
All reactions