kgriffs

import logging

import falcon
import falcon.asgi
import falcon.util


class HelloResource:
    async def on_get(self):
        pass

kgriffs

import asyncio
import multiprocessing

import httpx


class Publisher:
    def __init__(self):
        # NOTE(kgriffs): Explicitly manage the client so we can use the
        #   connection pool. This class could be made into a singleton

kgriffs

# Grant Jenks' LRU Cache with TTL for Python
#
# https://stackoverflow.com/questions/31771286/python-in-memory-cache-with-time-to-live/71634221#71634221

from functools import lru_cache, wraps
from time import monotonic


def lru_cache_with_ttl(maxsize=128, typed=False, ttl=60):
    """Least-recently used cache with time-to-live (ttl) limit."""

kgriffs

In [16]: data = {'some': 'data'}

In [17]: data_compressed = gzip.compress(json.dumps(data, ensure_ascii=False).encode())

In [18]: requests.post("https://httpbin.org/anything", files={"msg": ('message.json.gz', body_compressed)}).json()

Out[18]:
{'args': {},
 'data': '',
 'files': {'msg': 'data:application/octet-stream;base64,H4sIAGeLyGIC/6tWKs7PTVWyUlBKSSxJVKoFABmPLCMQAAAA'},

kgriffs

import gzip
import io

import lz4.frame


_DATA_CHUNKS = [
    '''
<div class="section" id="simple-usage">
<h2>Simple usage<a class="headerlink" href="#simple-usage" title="Permalink to this headline">¶</a></h2>

kgriffs

import falcon
import httpx


class TestResource:
    def on_get(self, req, resp):
        raise falcon.HTTPTooManyRequests()


def test_a_thing():

kgriffs

# Should match all non-printables except Unicode whitespace characters
r = re.compile(r'[^\w\s' + re.escape(string.punctuation) + ']')

def clean(text):
    return r.sub('', text).strip()

kgriffs

import asyncio
import contextvars
import random


var = contextvars.ContextVar('var', default=1776)

all_before = set()
all_after = set()

kgriffs

gAAAAABh9GdhynXgI1pQ-8ZVmquwrNvcrRQNDkcvh7ZKCWChPTF5h495d-890C3Kx9Ou-xoHZNUzvMXN_sUTse45q4J5mS5Xl2yb-3JTLbfo5hJZRsOHHmn2Bj4oVXtskCPiYo1W5beTrcNCLjo-wabGh4u9o-NSMc4Fp8wocIjP7ZTJSheBg4ZSw7Dxe_H7g9yzvpC2JSk8CNWAHeQCayC2xsYgJqMB957F1mTA3eL4Jv8UX-jqRVsGDecs6L1GPUaXjemExyOXcG4pNn9ys4dlEt2rIJYjLg==

kgriffs

# Courtesy of Vytautas Liuolia @vytas7 

import aiohttp
import falcon.asgi


class Proxy(object):
    UPSTREAM = 'https://falconframework.org'

    async def handle(self, req, resp):