BlockingPortalProvider fails on 2nd call with RuntimeError: Event loop is closed
#743
|
from __future__ import annotations
import trio
import anyio
from ipaddress import IPv4Address
import httpx
from anyio.from_thread import BlockingPortalProvider
class AsyncAPI:
_client: httpx.AsyncClient
def __init__(self):
self._client = httpx.AsyncClient()
@property
def sync(self) -> SyncAPI:
return SyncAPI(self)
async def get_ipaddress(self) -> IPv4Address:
url = "https://ifconfig.me/ip"
response = await self._client.get(url)
response.raise_for_status()
return IPv4Address(response.text)
class SyncAPI:
_portal_provider: BlockingPortalProvider
_async_api: AsyncAPI
def __init__(self, api: AsyncAPI):
self._portal_provider = BlockingPortalProvider()
self._async_api = api
def get_ipaddress(self) -> IPv4Address:
with self._portal_provider as portal:
return portal.call(self._async_api.get_ipaddress)
api = AsyncAPI()
print('#'*80)
print('# First Call')
print('#'*80)
ipaddress = api.sync.get_ipaddress()
print(repr(ipaddress))
print()
print('#'*80)
print('# Second Call')
print('#'*80)
ipaddress = api.sync.get_ipaddress()
print(repr(ipaddress)) |
Replies: 10 comments 28 replies
|
I've opened a discussion as it's likely user error, but I'm not sure what I'm doing wrong above? |
|
Note It does seem to work fine if I instead use |
|
Ideally, you should have the provider as a global, so that other instances of your classes can take advantage of it. If you have a private instance, then it may be pointless to use the provider vs just opening a regular blocking portal. |
|
Does this occur with any other third party library besides httpx? |
|
AFAICT this usage mirrors the example in the docs: from anyio.to_thread import BlockingPortalProvider
class MyAPI:
def __init__(self, async_obj) -> None:
self._async_obj = async_obj
self._portal_provider = BlockingPortalProvider()
def do_stuff(self) -> None:
with self._portal_provider as portal:
portal.call(async_obj.do_async_stuff)https://anyio.readthedocs.io/en/stable/threads.html#sharing-a-blocking-portal-on-demand |
|
If I make portal_provider = BlockingPortalProvider('asyncio')
class SyncAPI:
_async_api: AsyncAPI
def __init__(self, api: AsyncAPI):
self._async_api = api
def get_ipaddress(self) -> IPv4Address:
with portal_provider as portal:
return portal.call(self._async_api.get_ipaddress) |
|
storing a class SyncAPI:
_async_api: AsyncAPI
def __init__(self, api: AsyncAPI):
self._async_api = api
self._portal = portal_provider.__enter__()
def get_ipaddress(self) -> IPv4Address:
return self._portal.call(self._async_api.get_ipaddress)>>> api.sync.get_ipaddress()
IPv4Address('<redacted>')
>>> api.sync.get_ipaddress()
IPv4Address('<redacted>')
>>> api.sync.get_ipaddress()
IPv4Address('<redacted>')So, maybe I just need to manually create a |
|
It's hard for me to give proper advice as I don't quite understand the scope and purpose of the API you're trying to create. |
It's just a convenience method to return a In the actual app I use a from __future__ import annotations
from functools import cached_property
from ipaddress import IPv4Address
import anyio
import httpx
import trio
import wrapt
from anyio.from_thread import BlockingPortalProvider
portal_provider = BlockingPortalProvider('trio')
class AsyncAPI:
_client: httpx.AsyncClient
def __init__(self):
self._client = httpx.AsyncClient()
@cached_property
def sync(self) -> SyncAPI:
return SyncAPI(self)
async def get_ipaddress(self) -> IPv4Address:
url = "https://ifconfig.me/ip"
response = await self._client.get(url)
response.raise_for_status()
return IPv4Address(response.text)
class SyncAPI(wrapt.ObjectProxy):
def __init__(self, wrapped):
super().__init__(wrapped)
def get_ipaddress(self) -> IPv4Address:
with portal_provider as portal:
return portal.call(self.__wrapped__.get_ipaddress)>>> api = AsyncAPI()
>>> api.sync.get_ipaddress()
IPv4Address('<redacted>')
>>> api.sync.get_ipaddress()
IPv4Address('<redacted>') |
This will deeply involve how your application is structured. Let me just ask this: when would you ideally expect the async event loop to be terminated in your application lifecycle? |
|
For the sync apps, or in an interactive notebook session I'd want the event loop to only be closed at interpreter shutdown. I expect the api to be hit throughout the lifetime of sync apps based on different user inputs. The sync API will likely be used by less experienced programmers who would probably trip over global context managers. It will also be used interactively in notebooks by analysts so they'll want the ability to just run it in multiple different cells. |
|
Ok, so that then warrants the use of |
|
Thanks for your help @agronholm. I've gone ahead and marked this as answered since I now know why it's happening and how it could be worked around. For now I'll use the trio backend as it appears to work out of the box so is the path of least resistance. I'll make sure to come back and warn against it if I live to rue that decision! 😅 |
|
No, I just don't understand how it could make a difference. Sorry, I'm likely being dense, but are you saying he below is correct / will work? from ipaddress import IPv4Address
import httpx
from anyio.from_thread import BlockingPortalProvider
portal_provider = BlockingPortalProvider('trio')
class AsyncAPI:
_client: httpx.AsyncClient
def __init__(self):
self._client = httpx.AsyncClient()
async def get_ipaddress(self) -> IPv4Address:
url = "https://ifconfig.me/ip"
response = await self._client.get(url)
response.raise_for_status()
return IPv4Address(response.text)
class SyncAPI:
def __init__(self, async_api: AsyncAPI):
self._async_api = async_api
def get_ipaddress(self) -> IPv4Address:
with portal_provider as portal:
return portal.call(self._async_api.get_ipaddress)
async_api = AsyncAPI()
sync_api = SyncAPI(async_api) |
|
Well, the most obvious problem here is what |
|
I don't think I would ever use both apis in a single app. It would either be all sync (via the blocking portal) or it would be an async app, using only the async api. I can't imagine in an async app why you would want to use the sync api! |
|
If 1) |
|
The only time I've mixed the apis is testing in a Jupyter notebook but I'd be happy just saying "Don't do that"! Most analysts wouldn't know |
|
Thanks again for all your help understanding the nuances! It has given me a lot to think about! 😅 |
|
No problem, this really helped me to understand the practical challenges of this approach, and I'll be sure to incorporate it in my talk on Wednesday. |
|
xref: An example using |
|
Hello ! I seems this issue is the same as encode/httpcore#659 which snowballs to pydantic/pydantic-ai#748 Just wanted GitHub to keep a link between those related issues :) Also: if the same issue keeps beeing seen downstream, does it mean comething should change in anyio ?
I'm interested in an opinion of who did "wrong" (with all sorts of quotes) in the pydantic case according to anyio's knowledgeable folks: httpcore or pydantic ? (or the end user ? But I don't think so) |
|
From reading the issue it seems like they're not managing the connection pool correctly. |
|
Thanks a lot for weighing in :) 🙏 |
If 1)
httpx.AsyncClientdefers the code path selection to the first use of the APIs, and 2) you never use both async and sync APIs from the same process, then you should be fine.