tomchristie

import trio

async def main():
    async with ws_connect("ws://127.0.0.1:8765") as websockets:
        await websockets.send("Hello, world.")
        message = await websockets.recv()
        print(message)

trio.run(main)

tomchristie

class Client():
    ...

    def request(
        self,
        method: str,
        url: URLTypes,
        *,
        content: RequestContent = None,
        data: RequestData = None,

tomchristie

from contextlib import contextmanager


class Transport:
    @contextmanager
    def request(self, method, url, headers=None, stream=None):
        connection_closed = False

        def content():
            for chunk in [b"Hello", b", ", b"world", b"!"]:

tomchristie

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://login.microsoftonline.com/")
        print(r)

asyncio.run(main(), debug=True)

tomchristie

class ThrottleTransport:
    def __init__(self, throttle, **kwargs):
        self.pool = httpcore.SyncConnectionPool(**kwargs)
        self.history = []
        
        # Parse the thottle, which should be a string, like '100/minute'.
        count, _, duration = throttle.partition('/')
        self.max_in_history = int(count)
        self.cutoff = {'second': 1.0, 'minute': 60.0, 'hour': 3600.0}[duration]

tomchristie

Maintainer driven issues.

Hi there - we’re trying out something new on this repository. We’re calling it “maintainer driven” issues & pull requests.

What does this mean for you?

In general: Please do not open issues or pull requests on this repository.

Instead, we’d encourage you to instead talk through any issues or suggestions in the chat room, or on the discussion group. If the maintainance team would like to escalate a discussion into an issue then we may do so, or ask you to do so.

tomchristie

import httpx
from starlette.requests import Request
from starlette.responses import StreamingResponse


class Proxy:
    def __init__(self, hostname):
        self.hostname = hostname
        self.client = httpx.AsyncClient()

tomchristie

class LifespanManager:
    def __init__(self, app):
        self.app = app
        self.startup_complete = asyncio.Event()
        self.shutdown_complete = asyncio.Event()
        self.messages = [{'type': 'lifespan.startup'}, {'type': 'lifespan.shutdown'}]

    async def __aenter__(self):
        self.task = asyncio.create_task(self.app(self.receive, self.send))
        await self.startup_complete.wait()

tomchristie

HTTPCore

A proposal for requests III.

Feature support

• HTTP/1.1 and HTTP/2 Support.
• async/await support for non-thread-blocking HTTP requests.
• Fully type annotated.
• 98% test coverage.

tomchristie

import aiohttp

from django.views.generic import TemplateView


class FetchView(TemplateView):
    template_name = 'index.html'

    async def get(self, request, *args, **kwargs):
        async with aiohttp.ClientSession() as session: