Skip to main content

Python & FastAPI Fundamentals — Async Python, API endpoints, dependency injection

Created: December 12, 2025 Larry Qu 10 min read

Overview

Modern web APIs need to be fast, reliable, and easy to maintain. FastAPI (https://fastapi.tiangolo.com/) combines Python’s type hints with an async-first design to deliver high-performance, developer-friendly APIs. This guide covers three fundamental concepts you should master: asynchronous Python, API endpoint design, and FastAPI’s dependency injection system. Each section includes concise explanations, production-quality examples, and practical tips.


Async Python — what it is and why it matters

What async is

  • Asyncio (the async/await model) is Python’s approach for non-blocking concurrency. It’s ideal for I/O-bound workloads (HTTP calls, DB queries, file I/O) because it lets a single thread manage many concurrent tasks by yielding control on awaits.
  • Use async for handling many simultaneous requests without spawning many threads or processes; avoid blocking the event loop with CPU-heavy synchronous work.

Practical example — concurrent HTTP requests

The example below demonstrates an async endpoint that concurrently fetches multiple URLs using httpx.AsyncClient. It is runnable and shows how to structure async paths in FastAPI.

# content: examples/async_fetch.py
from typing import List
import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    resp = await client.get(url)
    resp.raise_for_status()
    return {"url": url, "status": resp.status_code}

@app.get("/fetch")
async def fetch_urls(urls: List[str] = []):
    """Fetch multiple URLs concurrently.

    Provide `?urls=https://a.com&urls=https://b.com` or a single `urls` query param.
    """
    if not urls:
        raise HTTPException(status_code=400, detail="urls parameter is required")

    # Limit concurrency to avoid overwhelming remote servers and connection pools
    sem = asyncio.Semaphore(10)

    async def bounded_fetch(u: str):
        async with sem:
            return await fetch(client, u)

    async with httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=20)) as client:
        tasks = [bounded_fetch(u) for u in urls]
        results = await asyncio.gather(*tasks)
    return {"results": results}

Notes:

  • Use async view functions for non-blocking I/O.
  • Prefer libraries with native async support (e.g., httpx, async DB drivers).
  • Avoid time.sleep() and other blocking calls; use await asyncio.sleep() instead.

Core terms & abbreviations

  • ASGI — Asynchronous Server Gateway Interface: the async successor to WSGI used by frameworks like FastAPI; servers include Uvicorn and Hypercorn. (ASGI spec)
  • Uvicorn — A fast ASGI server commonly used to run FastAPI applications. (https://www.uvicorn.org/)
  • Pydantic — Data validation & settings management library used by FastAPI for request/response models. (https://pydantic-docs.helpmanual.io/)
  • SQLModel — A library from the FastAPI author combining Pydantic & SQLAlchemy patterns for typing-friendly DB models. (https://sqlmodel.tiangolo.com/)
  • DI — Dependency Injection: a pattern to provide components (DB, clients, auth) via Depends in FastAPI.
  • JWT — JSON Web Token, a compact, URL-safe token for stateless auth. (https://jwt.io/)

Quick reference: ASGI, Uvicorn, Pydantic, SQLModel, DI (Dependency Injection), JWT.

Handling blocking (CPU-bound) work

If you must run blocking code (e.g., CPU-heavy tasks), offload it to a thread/process pool with asyncio.get_event_loop().run_in_executor or use a background job queue (Celery, RQ, or Prefect).

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def blocking_work(x: int) -> int:
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(executor, lambda: x ** 2)
    return result

Async pitfalls & best practices

  • Don’t call blocking I/O in async code (library calls using sync sockets, requests, time.sleep).
  • Keep the event loop responsive: avoid long-running sync loops.
  • Monitor event loop latency in production; alerts help catch blocking operations early.

API Endpoints — building robust, typed routes

What API endpoints are

  • FastAPI maps Python function signatures and type hints to request parsing, input validation, and OpenAPI docs automatically.
  • Use Pydantic models for request/response validation and response_model for consistent output shapes.

Example — users API with Pydantic

This small example shows common patterns: path params, query params, request body validation, response models, and proper status codes.

# content: examples/users_api.py
from typing import Optional, List
from pydantic import BaseModel, EmailStr
from fastapi import FastAPI, HTTPException, status, Query, Path

class UserIn(BaseModel):
    email: EmailStr
    password: str

class UserOut(BaseModel):
    id: int
    email: EmailStr

app = FastAPI()

# In-memory store for example purposes
_users: List[UserOut] = []

@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserIn):
    # NOTE: never store raw passwords in production; hash them with bcrypt/argon2
    user = UserOut(id=len(_users) + 1, email=payload.email)
    _users.append(user)
    return user

@app.get("/users", response_model=List[UserOut])
async def list_users(limit: int = Query(10, ge=1, le=100), offset: int = 0):
    return _users[offset : offset + limit]

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int = Path(..., gt=0)):
    try:
        return _users[user_id - 1]
    except IndexError:
        raise HTTPException(status_code=404, detail="User not found")

    ### Async DB example (SQLModel + async session)

    This example shows how to expose an async session as a dependency and query data using SQLModel (async SQLAlchemy engine).

    ```python
    from typing import AsyncGenerator
    from sqlmodel import SQLModel, Field, select
    from sqlmodel.ext.asyncio.session import AsyncSession
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine

    DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=False, future=True)

    class User(SQLModel, table=True):
        id: int | None = Field(default=None, primary_key=True)
        email: str

    async def init_db():
        async with engine.begin() as conn:
            await conn.run_sync(SQLModel.metadata.create_all)

    async def get_session() -> AsyncGenerator[AsyncSession, None]:
        async with AsyncSession(engine) as session:
            yield session

    @app.on_event("startup")
    async def on_startup():
        await init_db()

    @app.get("/db/users", response_model=List[User])
    async def db_list_users(session: AsyncSession = Depends(get_session)):
        result = await session.execute(select(User))
        users = result.scalars().all()
        return users
    ```

    Notes:
    - Use async DB drivers (e.g., `asyncpg` for Postgres) in production.
    - Ensure pool sizes and connection limits are configured to match concurrency needs.

Notes & patterns:

  • Always validate request bodies with Pydantic models (UserIn).
  • Use status constants and response_model to control output & docs.
  • Document constraints using Query, Path, and Body parameters.

Best practices

  • Validate input and sanitize outputs.
  • Keep functions small and pure; delegate business logic to service layers.
  • Use background tasks (BackgroundTasks) for non-critical work like emails.

Dependency Injection (DI) — clean, testable components

What dependency injection is

  • FastAPI’s DI uses Depends to declare dependencies in function signatures. Dependencies can be sync or async and can be used to provide services like DB sessions, authentication, configuration, or clients.
  • Dependencies improve testability (they can be overridden during tests) and separation of concerns.

Simple dependency example (auth)

# content: examples/deps_auth.py
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional

app = FastAPI()

async def get_current_user(authorization: Optional[str] = None):
    # In real code, read header via Header(...) or use OAuth2 schemes
    if authorization != "Bearer secrettoken":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return {"username": "alice"}

@app.get("/me")
async def me(user=Depends(get_current_user)):
    return {"user": user}

### OAuth2 (password flow) + JWT example

This is a concise example showing `OAuth2PasswordBearer`, a token endpoint, and a `get_current_user` dependency that validates a JWT-like token.

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from datetime import datetime, timedelta import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token") SECRET_KEY = “change-this-secret”

def create_token(data: dict, expires_delta: timedelta | None = None) -> str: to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) to_encode.update({“exp”: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=“HS256”)

@app.post(’/token’) async def token(form_data: OAuth2PasswordRequestForm = Depends()): # Validate user (demo only) if form_data.username != ‘alice’ or form_data.password != ‘wonderland’: raise HTTPException(status_code=400, detail=‘Incorrect username or password’) token = create_token({“sub”: form_data.username}) return {“access_token”: token, “token_type”: “bearer”}

async def get_current_user_token(token: str = Depends(oauth2_scheme)): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[“HS256”]) return {“username”: payload.get(‘sub’)} except jwt.PyJWTError: raise HTTPException(status_code=401, detail=‘Invalid token’)

@app.get(’/me2’) async def me2(user=Depends(get_current_user_token)): return {“user”: user}


Notes: In production use well-audited libs (e.g., `python-jose`, `Authlib`) and secure secret management.

### Background tasks

For non-critical work (emails, notifications, light processing) use `BackgroundTasks` to avoid blocking the request/response cycle:

from fastapi import BackgroundTasks

def send_welcome_email(email: str) -> None: # pretend to send an email (sync or delegate to worker) print(f"sending email to {email}")

@app.post(’/register’) async def register(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(send_welcome_email, email) return {“status”: “ok”}


### WebSockets

FastAPI supports WebSocket endpoints for bidirectional communication (e.g., live updates, chat).

from fastapi import WebSocket

@app.websocket(’/ws’) async def websocket_endpoint(ws: WebSocket): await ws.accept() try: while True: data = await ws.receive_text() await ws.send_text(f"echo: {data}") except Exception: await ws.close()


### Resource lifecycle: database session via yield (cleanup)

Use `contextlib.asynccontextmanager` or a `yield` dependency to create and clean up resources (database transactions, network clients).

content: examples/deps_db.py

from fastapi import FastAPI, Depends from typing import AsyncGenerator from contextlib import asynccontextmanager

app = FastAPI()

class DummyAsyncDB: async def connect(self): print(“connect”)

async def disconnect(self):
    print("disconnect")

async def fetch_user(self, user_id: int):
    return {"id": user_id, "email": "[email protected]"}

@asynccontextmanager async def get_db() -> AsyncGenerator[DummyAsyncDB, None]: db = DummyAsyncDB() await db.connect() try: yield db finally: await db.disconnect()

@app.get("/db/users/{user_id}") async def user_profile(user_id: int, db: DummyAsyncDB = Depends(get_db)): return await db.fetch_user(user_id)

Notes:

- `yield`-based deps ensure cleanup runs after the request completes, even on errors.
- Dependencies can be nested; FastAPI resolves and caches dependencies per-request by default.

### Testing & overrides

- Use `TestClient` and `app.dependency_overrides` to inject test doubles for dependencies (e.g., mocked DB), making tests deterministic and fast.

from fastapi.testclient import TestClient from examples.deps_db import app, get_db

client = TestClient(app)

async def fake_db(): class FakeDB: async def fetch_user(self, user_id): return {“id”: user_id, “email”: “[email protected]”} yield FakeDB()

app.dependency_overrides[get_db] = fake_db res = client.get(’/db/users/1’) assert res.json()[’email’] == ’[email protected]


### DI pitfalls & best practices

- Avoid storing mutable global state in dependencies unless it's readonly or thread-safe.
- Be mindful of per-request caching: by default, a dependency used multiple times in a single request is executed once and its value reused.
- Use `Depends` for cross-cutting concerns (auth, db session, settings) and keep handlers focused on request semantics.

---

## Putting it all together — a small, async FastAPI app

This example ties async work, endpoints, and DI together with a small service that fetches external data and uses a dependency-provided HTTP client.

content: examples/app_together.py

from typing import List import httpx from fastapi import FastAPI, Depends from contextlib import asynccontextmanager

app = FastAPI()

@asynccontextmanager async def get_http_client(): async with httpx.AsyncClient(timeout=10.0) as client: yield client

@app.get(’/multi-fetch’) async def multi_fetch(urls: List[str], client: httpx.AsyncClient = Depends(get_http_client)): tasks = [client.get(u) for u in urls] responses = await httpx.gather(*tasks) return {“statuses”: [r.status_code for r in responses]}


Run with:

pip install fastapi[all] httpx uvicorn examples.app_together:app –reload


---

## Deployment & architecture (text graphs)

- Simple deployment:

client(browser or mobile) -> CDN -> load balancer -> FastAPI (uvicorn workers) -> database


- Serverless / Functions approach:

git -> CI -> deploy to platform (Vercel, AWS Lambda, Cloudflare Workers) -> managed DB / storage


Deployment notes:

- Use ASGI servers (uvicorn, hypercorn) for FastAPI.
- For high throughput, run multiple worker processes (e.g., Gunicorn with Uvicorn workers) and tune connection pools for DB.
- Use observability: structured logs, metrics, tracing; track latency and errors.

## Tips for running in production

- Example Gunicorn + Uvicorn workers command:

gunicorn -k uvicorn.workers.UvicornWorker -w 4 examples.app_together:app


- A common heuristic for worker count: `workers = 2 * cpu_count + 1` for CPU-bound apps; tune for I/O-bound async apps based on performance testing.
- Configure connection pools for DB clients to match concurrency and avoid exhausting DB connections.

---

## Pros / Cons & alternatives

- Pros:
  - High performance for I/O-bound workloads (async-first).
  - Automatic validation & OpenAPI docs from type hints.
  - Easy testing and dependency isolation via `Depends`.

- Cons:
  - Async programming has an initial learning curve.
  - Careful with mixing blocking libraries; prefer async-native drivers.

- Alternatives: Flask (simple, sync-first), Django REST Framework (feature-rich, synchronous), FastAPI + SQLModel/Databases for type-safe DB access.
**Alternatives:** Flask (simple, sync-first), Django REST Framework (feature-rich, synchronous), FastAPI + SQLModel/Databases for type-safe DB access.

### Alternatives & when to choose them

- **Flask** — great for simple and small services or when an ecosystem of extensions is preferred; sync-first.
- **Django REST Framework** — full-featured, batteries-included for large projects needing auth, admin, and conventions.
- **Go (Gin/Fiber)** — consider for extremely high throughput and a compiled single-binary deployment model.
- **Serverless (Vercel, Cloudflare Workers)** — ideal for scaling short-lived functions with low operational overhead.

---

## Further resources

- FastAPI docs — <https://fastapi.tiangolo.com/>
- Pydantic — <https://pydantic-docs.helpmanual.io/>
- AsyncIO docs — <https://docs.python.org/3/library/asyncio.html>
- httpx — <https://www.python-httpx.org/>
- SQLModel (type-safe ORM for FastAPI) — <https://sqlmodel.tiangolo.com/>
- RealPython Asyncio Guide — <https://realpython.com/async-io-python/>

---

## Conclusion

FastAPI brings together modern Python language features (type hints and `async`/`await`) with pragmatic API tooling (Pydantic validation, automatic docs, and dependency injection). Mastering async patterns, designing typed endpoints, and using `Depends` for clean DI will let you build fast, maintainable, and testable APIs. If you'd like, I can scaffold a small starter repo (`starter/python-fastapi-basic`) with examples, test coverage, and a GitHub Actions workflow next.

## Resources

- [MDN Web Docs](https://developer.mozilla.org/)
- [Web.dev](https://web.dev/)
- [Can I Use](https://caniuse.com/)

Comments

Share this article

Scan to read on mobile

👍 Was this article helpful?