Skip to content

Instantly share code, notes, and snippets.

View demo.py
import asyncio
import contextlib
async def child(sock):
try:
with contextlib.closing(sock):
await asyncio.sleep(2)
except Exception as e:
print(f"child task got error: {type(e)=} {e=}")
raise
View ressurect_coro.py
import objgraph
import asyncio
READY = object()
import weakref
import asyncio
import gc
import logging
@graingert
graingert / demo.py
Created Jul 31, 2022
running trio in asyncio and getting the current task context
View demo.py
import asyncio
import collections.abc
import contextlib
import functools
import types
import httpx
import sniffio
import trio
View trio_in_asyncio.py
import asyncio
import collections.abc
import contextlib
import functools
import types
import httpx
import sniffio
import trio
View become_twisted.py
from twisted.internet import defer
import curio
import concurrent
class Interceptor:
def __init__(self, send, coro):
self._send = send
self._coro = coro
View contextvar_runner.py
__all__ = 'run',
import functools
import asyncio
from asyncio import coroutines
from asyncio import events
from asyncio import tasks
import contextlib
@contextlib.contextmanager
@graingert
graingert / .gitignore
Last active Jul 28, 2022
twisted asyncio tests
View .gitignore
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
View async_gen_handle_cancel.py
import asyncio
import contextlib
class _AcknowledgeException:
def __init__(self, exception):
self.exception = exception
class _RecommendGeneratorExit(BaseException):
@graingert
graingert / agen_close.py
Last active Jul 22, 2022
closing an async_generator_asend does not throw GeneratorExit into the `await` statement
View agen_close.py
import types
def underline(text):
last_line = text.splitlines()[-1]
return text + "\n" + "^" * len(last_line)
@types.coroutine
def _async_yield(v):
return (yield v)
View async_timeout.py
import asyncio
import async_timeout
import quattro
import contextlib
import anyio
async def main(note, cmgr):
try:
async with cmgr(-11):