ndugram/fasthttp

Declare a Pydantic model and pass it as response_model. FastHTTP will validate and parse the response automatically:

from fasthttp import FastHTTP
from fasthttp.response import Response
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str
    email: str


app = FastHTTP()


@app.get(
    url="https://jsonplaceholder.typicode.com/users/1",
    response_model=User,
)
async def get_user(resp: Response) -> User:
    return User(**resp.json())


if __name__ == "__main__":
    app.run()

Register as many routes as you need across all HTTP methods. FastHTTP runs them concurrently:

from fasthttp import FastHTTP
from fasthttp.response import Response

app = FastHTTP()


@app.get(url="https://httpbin.org/get")
async def get_data(resp: Response) -> dict:
    return resp.json()


@app.post(url="https://httpbin.org/post")
async def post_data(resp: Response) -> dict:
    return resp.json()


@app.put(url="https://httpbin.org/put")
async def put_data(resp: Response) -> dict:
    return resp.json()


@app.patch(url="https://httpbin.org/patch")
async def patch_data(resp: Response) -> dict:
    return resp.json()


@app.delete(url="https://httpbin.org/delete")
async def delete_data(resp: Response) -> int:
    return resp.status_code


@app.head(url="https://httpbin.org/get")
async def head_data(resp: Response) -> int:
    return resp.status


@app.options(url="https://httpbin.org/get")
async def options_data(resp: Response) -> dict:
    return {"allow": resp.headers.get("allow", "")}


if __name__ == "__main__":
    app.run()

Group related routes into a Router with a shared prefix or base URL, then include it into the app:

from fasthttp import FastHTTP, Router
from fasthttp.response import Response

users_router = Router(prefix="https://jsonplaceholder.typicode.com")


@users_router.get(url="/users/1")
async def get_user(resp: Response) -> dict:
    return resp.json()


@users_router.get(url="/users/2")
async def get_user_two(resp: Response) -> dict:
    return resp.json()


@users_router.post(url="/users")
async def create_user(resp: Response) -> dict:
    return resp.json()


app = FastHTTP()
app.include_router(users_router)

if __name__ == "__main__":
    app.run()

Intercept and modify requests before they are sent and responses after they are received:

from fasthttp import FastHTTP
from fasthttp.middleware import BaseMiddleware
from fasthttp.response import Response


class LoggingMiddleware(BaseMiddleware):
    __priority__ = 0
    __methods__ = None
    __enabled__ = True

    async def request(self, method: str, url: str, kwargs: dict) -> dict:
        print(f"→ {method} {url}")
        return kwargs

    async def response(self, response: Response) -> Response:
        print(f"← {response.status}")
        return response


app = FastHTTP(middleware=[LoggingMiddleware()])


@app.get(url="https://httpbin.org/get")
async def get_data(resp: Response) -> dict:
    return resp.json()


if __name__ == "__main__":
    app.run()

Use Depends to share logic across routes — auth tokens, computed headers, or any reusable setup:

from fasthttp import FastHTTP, Depends
from fasthttp.response import Response
from fasthttp.types import RequestsOptinal


def auth_headers() -> RequestsOptinal:
    return {"headers": {"Authorization": "Bearer my-token"}}


app = FastHTTP()


@app.get(
    url="https://httpbin.org/get",
    dependencies=[Depends(auth_headers)],
)
async def get_data(resp: Response) -> dict:
    return resp.json()


if __name__ == "__main__":
    app.run()

Run setup and teardown logic around your requests using an async context manager:

from contextlib import asynccontextmanager

from fasthttp import FastHTTP
from fasthttp.response import Response


@asynccontextmanager
async def lifespan(app: FastHTTP):
    print("Startup: loading credentials...")
    app.token = "my-secret-token"  # type: ignore[attr-defined]
    yield
    print("Shutdown: cleanup done.")


app = FastHTTP(lifespan=lifespan)


@app.get(url="https://httpbin.org/get")
async def get_data(resp: Response) -> dict:
    return resp.json()


if __name__ == "__main__":
    app.run()

By default all routes run fully in parallel. Use concurrency to cap how many requests execute at the same time — useful when the target API has a rate limit or you want predictable resource usage:

from fasthttp import FastHTTP
from fasthttp.response import Response

# At most 3 requests run simultaneously
app = FastHTTP(concurrency=3)


@app.get(url="https://api.example.com/items/1")
async def item_1(resp: Response) -> dict:
    return resp.json()


@app.get(url="https://api.example.com/items/2")
async def item_2(resp: Response) -> dict:
    return resp.json()


@app.get(url="https://api.example.com/items/3")
async def item_3(resp: Response) -> dict:
    return resp.json()


@app.get(url="https://api.example.com/items/4")
async def item_4(resp: Response) -> dict:
    return resp.json()


if __name__ == "__main__":
    # First 3 start immediately, item_4 waits for a free slot
    app.run()

Use @app.ws() to connect to a WebSocket endpoint and exchange messages in real time:

from fasthttp import FastHTTP, WebSocket

app = FastHTTP()


@app.ws(url="wss://echo.websocket.org")
async def echo(ws: WebSocket) -> None:
    await ws.send("Hello from fasthttp!")
    msg = await ws.recv()
    print(f"Received: {msg}")


if __name__ == "__main__":
    app.run()

Supports auto-reconnect with exponential backoff, async for streaming, and works alongside HTTP routes in the same app.run().

Use @app.graphql to send queries and mutations. The handler returns the query body; FastHTTP sends it and gives you the parsed response:

from fasthttp import FastHTTP
from fasthttp.response import Response


app = FastHTTP()


@app.graphql(url="https://countries.trevorblades.com/graphql")
async def get_countries(resp: Response) -> dict:
    return {
        "query": """
            {
                countries {
                    name
                    code
                    capital
                }
            }
        """
    }


if __name__ == "__main__":
    app.run()