Async HTTP Client

import asyncio
import contextlib
import time

from tornado import gen, httpclient, httpserver, ioloop, netutil, web
from tornado.escape import json_decode, json_encode


class CoverHandler(web.RequestHandler):
    async def get(self, isbn):
        try:
            default_delay = "3" if isbn.startswith("slow") else "0.05"
            await gen.sleep(float(self.get_query_argument("delay", default_delay)))
        except asyncio.CancelledError:
            return  # the client gave up: nothing left to answer
        if isbn.startswith("bad"):
            raise web.HTTPError(502, reason="upstream failure")
        self.write({"isbn": isbn, "cover_url": f"https://covers.example/{isbn}.jpg"})


class StatusHandler(web.RequestHandler):
    def get(self, code):
        self.set_status(int(code))
        self.write({"status": int(code)})


class MovedHandler(web.RequestHandler):
    def get(self):
        self.redirect("/covers/moved-978-0")


class EchoHeadersHandler(web.RequestHandler):
    def get(self):
        self.write({k.lower(): v for k, v in self.request.headers.get_all()})


class ChunkHandler(web.RequestHandler):
    async def get(self):
        self.set_header("X-Total-Chunks", "3")
        for i in range(3):
            self.write(f"chunk-{i};")
            await self.flush()


class UploadHandler(web.RequestHandler):
    def post(self):
        self.write({"received": len(self.request.body), "type": self.request.headers.get("Content-Type")})


def make_upstream():
    return web.Application(
        [
            (r"/covers/([^/]+)", CoverHandler),
            (r"/status/(\d+)", StatusHandler),
            (r"/moved", MovedHandler),
            (r"/headers", EchoHeadersHandler),
            (r"/chunks", ChunkHandler),
            (r"/upload", UploadHandler),
        ]
    )


@contextlib.asynccontextmanager
async def fake_upstream():
    sockets = netutil.bind_sockets(0, address="127.0.0.1")
    server = httpserver.HTTPServer(make_upstream())
    server.add_sockets(sockets)
    port = sockets[0].getsockname()[1]
    try:
        yield f"http://127.0.0.1:{port}"
    finally:
        server.stop()
        await server.close_all_connections()

async def demo_basic():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()
        response = await client.fetch(f"{base}/covers/978-0-13-468599-1")
        data = json_decode(response.body)
        print(response.code, response.reason, data["cover_url"])
        print(response.headers["Content-Type"], response.effective_url == f"{base}/covers/978-0-13-468599-1")
        print("fast enough:", response.request_time < 1.0)

        # string URL + kwargs, or an HTTPRequest object (but not both)
        request = httpclient.HTTPRequest(f"{base}/headers", headers={"X-Request-Id": "req-42"})
        echoed = json_decode((await client.fetch(request)).body)
        print(echoed["x-request-id"])
        try:
            await client.fetch(request, request_timeout=1)
        except ValueError as exc:
            print("ValueError:", exc)


asyncio.run(demo_basic())

async def demo_singleton():
    a = httpclient.AsyncHTTPClient()
    b = httpclient.AsyncHTTPClient()
    private = httpclient.AsyncHTTPClient(force_instance=True, defaults=dict(user_agent="bookshelf-batch/1.0"))
    print(a is b, a is private, type(a).__name__)
    print(a.defaults["request_timeout"], private.defaults["user_agent"])
    private.close()


asyncio.run(demo_singleton())

httpclient.AsyncHTTPClient.configure(
    None,  # SimpleAsyncHTTPClient, the default
    max_clients=2,
    defaults=dict(request_timeout=5.0, user_agent="bookshelf-live/1.0"),
)


async def demo_defaults_and_limit():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()
        print(client.max_clients, client.defaults["request_timeout"])
        sent = json_decode((await client.fetch(f"{base}/headers")).body)
        print(sent["user-agent"])
        sent = json_decode((await client.fetch(f"{base}/headers", user_agent="one-off/2.0")).body)
        print(sent["user-agent"])

        # four requests of about 0.3 s each, but only two run at a time
        started = time.perf_counter()
        responses = await asyncio.gather(
            *(client.fetch(f"{base}/covers/book-{i}?delay=0.3") for i in range(4))
        )
        elapsed = time.perf_counter() - started
        print(f"4 fetches with max_clients=2 took {elapsed:.1f}s")
        print("each request itself took", [round(r.request_time, 1) for r in responses], "(queueing excluded)")


asyncio.run(demo_defaults_and_limit())

async def demo_errors():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()

        try:
            await client.fetch(f"{base}/status/404")
        except httpclient.HTTPClientError as exc:
            print(exc, "| code", exc.code, "| body", exc.response.body)

        response = await client.fetch(f"{base}/status/503", raise_error=False)
        print(response.code, type(response.error).__name__, response.error)

        try:
            await client.fetch(f"{base}/covers/slow?delay=2", request_timeout=0.2, raise_error=False)
        except httpclient.HTTPClientError as exc:
            print("timeout:", type(exc).__name__, exc.code, exc.response)

        try:
            await client.fetch("http://127.0.0.1:1/")
        except OSError as exc:
            print("transport:", type(exc).__name__)


asyncio.run(demo_errors())

import base64
import secrets


async def demo_request_options():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()

        followed = await client.fetch(f"{base}/moved")
        print(followed.code, followed.effective_url.replace(base, ""))

        raw = await client.fetch(f"{base}/moved", follow_redirects=False, raise_error=False)
        print(raw.code, raw.headers["Location"])

        # Basic authentication sends an Authorization header
        password = secrets.token_hex(8)  # in production: read it from the environment or a secrets store
        sent = json_decode(
            (await client.fetch(f"{base}/headers", auth_username="bookshelf", auth_password=password)).body
        )
        print(base64.b64decode(sent["authorization"].split()[1]).decode().split(":")[0])

        # POST a JSON document
        reply = await client.fetch(
            f"{base}/upload",
            method="POST",
            headers={"Content-Type": "application/json"},
            body=json_encode({"event": "loan.created", "isbn": "978-0-13-468599-1"}),
        )
        print(json_decode(reply.body))


asyncio.run(demo_request_options())

import ssl


def tls_request_for_internal_api():
    context = ssl.create_default_context()  # verifies against the system trust store
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    # context.load_verify_locations(cafile="/etc/bookshelf/internal-ca.pem")
    # context.load_cert_chain("/etc/bookshelf/client.pem", "/etc/bookshelf/client.key")
    return httpclient.HTTPRequest("https://covers.internal.example/v1/covers/1", ssl_options=context)


request = tls_request_for_internal_api()
print(request.url, type(request.ssl_options).__name__)

async def demo_streaming_download():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()
        chunks, header_lines = [], []
        response = await client.fetch(
            f"{base}/chunks",
            streaming_callback=chunks.append,
            header_callback=header_lines.append,
        )
        print(b"".join(chunks).decode(), len(response.body))
        print(header_lines[0].strip(), "|", [h.strip() for h in header_lines if h.startswith("X-Total")])


asyncio.run(demo_streaming_download())

async def demo_streaming_upload():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient()
        pieces = [b"%PDF-1.7 " + b"x" * 1000 for _ in range(3)]

        async def producer(write):
            for piece in pieces:
                await write(piece)  # awaiting gives flow control

        reply = await client.fetch(
            f"{base}/upload",
            method="POST",
            headers={"Content-Type": "application/pdf", "Content-Length": str(sum(map(len, pieces)))},
            body_producer=producer,
        )
        print(json_decode(reply.body))


asyncio.run(demo_streaming_upload())

from tornado import locks


async def fetch_cover(client, base, isbn):
    response = await client.fetch(f"{base}/covers/{isbn}", request_timeout=2)
    return json_decode(response.body)["cover_url"]


async def demo_gather():
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient(force_instance=True, max_clients=10)
        isbns = ["978-0-13-468599-1", "bad-978-0-00", "978-1-49-195-0"]

        started = time.perf_counter()
        results = await asyncio.gather(*(fetch_cover(client, base, i) for i in isbns), return_exceptions=True)
        print(f"{time.perf_counter() - started:.2f}s", [r if isinstance(r, str) else f"error {r.code}" for r in results])

        # bound a large batch to 3 simultaneous upstream calls
        gate = locks.Semaphore(3)

        async def limited(isbn):
            async with gate:
                return await fetch_cover(client, base, isbn)

        covers = await asyncio.gather(*(limited(f"book-{n}") for n in range(12)))
        print(len(covers), "covers fetched")
        client.close()


asyncio.run(demo_gather())

import unittest

from tornado.testing import AsyncHTTPTestCase


class BookCoverHandler(web.RequestHandler):
    async def get(self, isbn):
        upstream = self.settings["covers_url"]
        client = httpclient.AsyncHTTPClient()
        try:
            response = await client.fetch(f"{upstream}/covers/{isbn}", request_timeout=0.5)
        except httpclient.HTTPClientError as exc:
            if exc.code == 599:
                raise web.HTTPError(504, reason="cover service timed out") from exc
            raise web.HTTPError(502, reason="cover service failed") from exc
        except OSError as exc:
            raise web.HTTPError(502, reason="cover service unreachable") from exc
        self.write({"isbn": isbn, "cover": json_decode(response.body)["cover_url"]})


class BookCoverTest(AsyncHTTPTestCase):
    def get_app(self):
        # the fake upstream is served by the same test server, under /up
        app = web.Application(
            [(r"/books/([^/]+)/cover", BookCoverHandler), (r"/up/covers/([^/]+)", CoverHandler)],
            covers_url=f"http://127.0.0.1:{self.get_http_port()}/up",
        )
        return app

    def test_cover(self):
        response = self.fetch("/books/978-0-13-468599-1/cover")
        self.assertEqual(response.code, 200)
        self.assertEqual(json_decode(response.body)["isbn"], "978-0-13-468599-1")

    def test_upstream_error_is_502(self):
        self.assertEqual(self.fetch("/books/bad-1/cover").code, 502)

    def test_upstream_timeout_is_504(self):
        self.assertEqual(self.fetch("/books/slow-1/cover").code, 504)


import logging

# the tests provoke 502 and 504 on purpose: keep the log quiet
logging.getLogger("tornado.access").setLevel(logging.CRITICAL)
logging.getLogger("tornado.application").setLevel(logging.CRITICAL)
suite = unittest.defaultTestLoader.loadTestsFromTestCase(BookCoverTest)
result = unittest.TextTestRunner(verbosity=0).run(suite)
assert result.wasSuccessful()

# norun -- needs the optional pycurl package (pip install pycurl) and a real proxy
from tornado import httpclient

httpclient.AsyncHTTPClient.configure(
    "tornado.curl_httpclient.CurlAsyncHTTPClient",
    max_clients=20,
    defaults=dict(proxy_host="proxy.internal.example", proxy_port=3128),
)


async def fetch_through_proxy():
    client = httpclient.AsyncHTTPClient()  # a CurlAsyncHTTPClient
    # per-request override, with proxy credentials read from the environment
    return await client.fetch("https://covers.example/v1/isbn/1", proxy_host="proxy.internal.example", proxy_port=3128)

async def demo_proxy_on_simple_client():
    httpclient.AsyncHTTPClient.configure(None)
    async with fake_upstream() as base:
        client = httpclient.AsyncHTTPClient(force_instance=True)
        try:
            await client.fetch(f"{base}/headers", proxy_host="127.0.0.1", proxy_port=3128)
        except NotImplementedError as exc:
            print("NotImplementedError:", exc)
        client.close()


asyncio.run(demo_proxy_on_simple_client())

import threading


def run_upstream_in_thread():
    """Run the fake upstream on its own loop in a worker thread; return (base_url, stop)."""
    ready = threading.Event()
    state = {}

    def target():
        async def main():
            async with fake_upstream() as base:
                state["base"], state["stop"] = base, asyncio.Event()
                state["loop"] = asyncio.get_running_loop()
                ready.set()
                await state["stop"].wait()

        asyncio.run(main())

    thread = threading.Thread(target=target, daemon=True)
    thread.start()
    ready.wait(5)
    return state["base"], lambda: (state["loop"].call_soon_threadsafe(state["stop"].set), thread.join(5))


base, stop = run_upstream_in_thread()
sync_client = httpclient.HTTPClient()
try:
    print(sync_client.fetch(f"{base}/covers/978-0-13-468599-1").code)
    try:
        sync_client.fetch(f"{base}/status/404")
    except httpclient.HTTPClientError as exc:
        print("raised:", exc)
finally:
    sync_client.close()
    stop()


async def sync_client_inside_a_loop():
    try:
        httpclient.HTTPClient()
    except RuntimeError as exc:
        print("RuntimeError:", exc)


asyncio.run(sync_client_inside_a_loop())

python -m tornado.httpclient --print_headers --print_body=false http://localhost:8888/books/1/cover

import ipaddress
import socket
from urllib.parse import urlsplit

ALLOWED_HOSTS = {"covers.example", "api.covers.example"}


def check_outgoing_url(url):
    parts = urlsplit(url)
    if parts.scheme != "https" or parts.hostname not in ALLOWED_HOSTS:
        raise ValueError(f"host not allowed: {parts.hostname}")
    return url


def is_public_address(host):
    for info in socket.getaddrinfo(host, None):
        address = ipaddress.ip_address(info[4][0])
        if address.is_private or address.is_loopback or address.is_link_local or address.is_reserved:
            return False
    return True


for candidate in ("https://covers.example/1.jpg", "https://169.254.169.254/latest/meta-data", "http://covers.example/1.jpg"):
    try:
        print("ok:", check_outgoing_url(candidate))
    except ValueError as exc:
        print("refused:", exc)
print(is_public_address("127.0.0.1"))

# norun -- needs the optional httpx package (pip install httpx)
import httpx
from tornado import web


class BookCoverHandler(web.RequestHandler):
    async def get(self, isbn):
        client: httpx.AsyncClient = self.settings["http"]  # one pooled client, created at start-up
        response = await client.get(f"/covers/{isbn}", timeout=2.0)
        response.raise_for_status()
        self.write(response.json())