Skip to content

Instantly share code, notes, and snippets.

@rafaelcaricio
Created July 8, 2021 09:00
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save rafaelcaricio/7930a653ca8e8de1e244d5546820028d to your computer and use it in GitHub Desktop.
Save rafaelcaricio/7930a653ca8e8de1e244d5546820028d to your computer and use it in GitHub Desktop.
Why not just use async Python for this API?
from fastapi import FastAPI
import requests
import httpx
import uuid
import logging
logging.basicConfig(format="%(asctime)s %(message)s")
log = logging.getLogger("myapp")
log.setLevel(logging.DEBUG)
app = FastAPI()
EXTERNAL_API_ENDPOINT = "http://localhost:8888"
@app.get("/healthcheck")
async def healthcheck():
return {"status": "ok"}
#
# Async mixed with blocking
#
def internal_signing_op(op_num: int, request_id: str) -> None:
session = requests.Session()
response = session.request("GET", EXTERNAL_API_ENDPOINT, timeout=2000)
log.debug(f"{request_id} {op_num}: {response}")
def sign_op1(request_id: str) -> None:
internal_signing_op(1, request_id)
def sign_op2(request_id: str) -> None:
internal_signing_op(2, request_id)
@app.get("/async-blocking")
async def root():
request_id = str(uuid.uuid4())
log.debug(f"{request_id}: started processing")
sign_op1(request_id)
sign_op2(request_id)
log.debug(f"{request_id}: finished!")
return {"message": "hello world"}
#
# Async end-to-end
#
async def v2_internal_signing_op(op_num: int, request_id: str) -> None:
"""Calls external API endpoint and returns the response as a dict."""
async with httpx.AsyncClient() as session:
response = await session.request("GET", EXTERNAL_API_ENDPOINT, timeout=2000)
log.debug(f"{request_id} {op_num}: {response}")
async def v2_sign_op1(request_id: str) -> None:
await v2_internal_signing_op(1, request_id)
async def v2_sign_op2(request_id: str) -> None:
await v2_internal_signing_op(2, request_id)
@app.get("/all-async")
async def v2_root():
request_id = str(uuid.uuid4())
log.debug(f"{request_id}: started processing")
await v2_sign_op1(request_id)
await v2_sign_op2(request_id)
log.debug(f"{request_id}: finished!")
return {"message": "hello world"}
import asyncio
from fastapi import FastAPI
import time
app = FastAPI()
@app.get("/")
async def root():
await asyncio.sleep(20)
return {"message": "Hello World"}
fastapi[all]==0.65.1
uvicorn[standard]==0.13.4
requests==2.25.1
httpx==0.18.2
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment