If the user cares about error reporting they should open an async context manager for you to run a nursery to handle that in
Yeah there should always be a place where a user of @madkinsz: library can introduce an async with above their cancellation
I prefer setting a reasonable timeout with
move_on_after(..., shield=True):
no success :(
it's a mess, don't judge xD
Is there a reason TaskGroup.start_soon is annotated to require a Coroutine and not accept a general Awaitable (asking because I want to schedule an async generators aclose method which is an awaitable and get type complaints (it seems to work in this case and I can cast it to fix the "error")
Or rather it needs a Callable returning a Coroutine but couldn't that still be loosened to an Awaitable?
8 replies
in master we accept any awaitables
With positional only arguments, you could do a more elegant interface, but those were added later. https://peps.python.org/pep-0570/
@nilueps you could just use a lambda
thanks
any interest in a "BufferedTextReceiveStream", basically an analog to "BufferedByteReceiveStream" but for strings? i need this for my app, but thought maybe I could make an anyio MR if useful
s, r = anyio.create_memory_object_stream(float('inf'), item_type=str)
bt = BufferedTextReceiveStream(r)
with s, r:
await s.send("foo")
assert await bt.receive() == "foo"
await s.send("bar")
assert await bt.receive_exactly(1) == "b"
assert await bt.receive_exactly(2) == "ar"
await s.send("baz")
assert await bt.receive_until("a", 100) == "b"
assert await bt.receive_exactly(1) == "z"
pytest-asyncio has
unused_tcp_port fixture. Which I'm not sure why it exists there... But is it interesting for AnyIO to have it as well?
I think it doesn't make sense for pytest-asyncio to have it, but as it have, does it make sense for anyio to have it as well?
(has*)
so as not to conflict with pytest-asyncio
that thing is trivial to implement
I think even anyio's test suite uses something like that internally
I can also start creating an issue 👀