Fast API

Installation

# Install
pip install "fastapi[standard]"
# Develop
fastapi dev main.py
# Production
fastapi run main.py

Feature

  • the interactive API documentation will be generated automatically

  • Uvicorn is used as a web server for Python

Concurrency

  • Starlette (and FastAPI) are based on AnyIO, which makes it compatible with both Python's standard library asyncio and Trio and provides an event loop similar to Node.js.

  • FastAPI applications run on ASGI servers like Uvicorn or Hypercorn. ASGI is an interface which allows communication between web server and Asynchronous web application and frameworks

@app.get("/test")
async def test():
    test = await fn()
    return test

Error Handling

app = FastAPI()
# Customize the http exception behavior and return the original func
@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)

# Customize the validation error of the body/query/param
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(request, exc)

@app.get("/test/{var}")
def test(var:str):
    if var == "invalid":
        raise HTTPException(status_code=404, detail="invalid value")
    return {"message": "Hello World"}

Middleware

from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import PlainTextResponse

# Can be used customize the request and response
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    print(f"Processing request: {request.url}")
    # The actual handler of request (api path)
    response = await call_next(request)
    print(f"Processed request: {response}")
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response
    


class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("Authorization")
        if not token or token != "Bearer valid-token":
            return PlainTextResponse(status_code=401, content="Unauthorized")
        return await call_next(request)

app = FastAPI()
app.add_middleware(AuthMiddleware)

@app.get("/secure-data/")
async def secure_data():
    return {"message": "This is secured data"}

Sub Applications

  • If you need to have two independent FastAPI applications, with their own independent OpenAPI and their own docs UIs, you can have a main app and "mount" one (or more) sub-application(s).

  • The middleware and error handler will also be independent between sub applications

app = FastAPI()
# define the sub application and its context path
# /app1/docs will be also be generated automatically
app1 = FastAPI(openapi_prefix="/app1")
# attach to the main application
app.mount("/app1", app1) 
# the api can be accessed through /app1/test
@app1.get("/test")
def test():
    return "app1 test"

Mounting

  • "Mounting" means adding a completely "independent" application in a specific path, that then takes care of handling everything under that path

from fastapi import FastAPI

app = FastAPI()


@app.get("/app")
def read_main():
    return {"message": "Hello World from main app"}


subapi = FastAPI()


@subapi.get("/sub")
def read_sub():
    return {"message": "Hello World from sub API"}


app.mount("/subapi", subapi)

Last updated

Was this helpful?