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

vskrachkov

from typing import Callable, Union


class Pipe:
    def __init__(self, callable_: Callable):
        self.callable_ = callable_

    def __rshift__(self, then: Union[Callable, "Pipe"]) -> "Pipe":
        if not isinstance(then, Pipe):
            then = Pipe(then)

dabeaz

# aproducer.py
#
# Async Producer-consumer problem.
# Challenge: How to implement the same functionality, but no threads.

import time
from collections import deque
import heapq

class Scheduler:

bobbytables

#!/usr/bin/env bash
# This script is meant to build and compile every protocolbuffer for each
# service declared in this repository (as defined by sub-directories).
# It compiles using docker containers based on Namely's protoc image
# seen here: https://github.com/namely/docker-protoc

set -e

REPOPATH=${REPOPATH-/opt/protolangs}
CURRENT_BRANCH=${CIRCLE_BRANCH-"branch-not-available"}