graingert

import asyncio
import collections.abc
import contextlib
import functools
import types

import httpx
import sniffio
import trio

graingert

import asyncio
import collections.abc
import contextlib
import functools
import types

import httpx
import sniffio
import trio

graingert

from twisted.internet import defer
import curio
import concurrent


class Interceptor:
    def __init__(self, send, coro):
        self._send = send
        self._coro = coro

graingert

__all__ = 'run',

import functools
import asyncio
from asyncio import coroutines
from asyncio import events
from asyncio import tasks
import contextlib

@contextlib.contextmanager

graingert

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

# C extensions
*.so

# Distribution / packaging
.Python

graingert

import asyncio
import contextlib


class _AcknowledgeException:
    def __init__(self, exception):
        self.exception = exception


class _RecommendGeneratorExit(BaseException):

graingert

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)

graingert

import asyncio
import async_timeout
import quattro
import contextlib
import anyio


async def main(note, cmgr):
    try:
        async with cmgr(-11):

graingert

import asyncio

async def main():
    try:
        async with asyncio.timeout(0):
            await asyncio.sleep(0)
    except TimeoutError:
        print("PASS: great a timeout is here!")
    else:
        print("FAILED: expected a timeout")

graingert

import psutil
import multiprocessing
import socket
import sys
import concurrent.futures


ctx = multiprocessing.get_context("spawn")

def target(sock):