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 AllowPUTAsCreateMixin(object):
    """
    The following mixin class may be used in order to support PUT-as-create
    behavior for incoming requests.
    """
    def update(self, request, *args, **kwargs):
        partial = kwargs.pop('partial', False)
        instance = self.get_object_or_none()
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)

tomchristie

Visa requirements for UK conferences.

First you'll want to check if you need a visa to enter the UK.

• Nationals of the EU, Switzerland, and EEA countries will not need a visa, and are free to work and conduct business in the UK.
• Nationals from some designated countries such as USA may travel for tourist or business purposes under a visa exemption. The link above will detail what documents you need to provide in order to travel under a visa exemption.
• Other nationals will need a visa.

The rules for allowable business travel under a visa exemption are covered by the same rules as those travelling under a vistor visa. If you will be travelling under a visa-exemption and the conference is covering your costs in any way you should make sure to read the relevant section below.

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 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

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

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.