SoFunction
Updated on 2025-04-06

How to build complex web APIs through FastAPI in Python

Build complex web APIs with FastAPI

Building complex Web APIs usually involve multiple aspects, including good architectural design, clear routing organization, data verification and processing, security measures, performance optimization, etc. The key to building complex web APIs with FastAPI is to take advantage of the tools and features it provides while following software engineering best practices. Here is a detailed guide to help you build a complex and efficient web API using FastAPI.

1. Project Structure

Create a reasonable folder structure for your project, which helps with the organization and maintenance of your code. For example:

my_project/
├── app/
│   ├── __init__.py
│   ├── 
│   ├── 
│   ├── 
│   ├── 
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── 
│   │   └── 
│   ├── services/
│   │   ├── __init__.py
│   │   ├── item_service.py
│   │   └── user_service.py
│   ├── 
│   └── 
└── tests/
    ├── __init__.py
    └── test_main.py

2. Dependency management

useOr more modernpoetryTo manage project dependencies. Ensure that all libraries required for development and production environments are listed.

# Use poetry to initialize the project and add dependenciespoetry init
poetry add fastapi uvicorn sqlalchemy alembic bcrypt passlib[bcrypt] pydantic email-validator

3. Configuration Management

Use environment variables or configuration files to manage different settings of your application (such as database connection strings, keys, etc.). You can use Pydantic'sBaseSettingskind.

# 
from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My Complex API"
    admin_email: str
    items_per_user: int = 50
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

4. Database integration

Select the right ORM (such as SQLAlchemy) and integrate with FastAPI. Create a database model and consider using Alembic for database migration.

# 
from sqlalchemy import create_engine
from  import declarative_base
from  import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        ()

5. Define the model and pattern

Create data models for mapping to database tables, and Pydantic models for verification of request bodies and responses.

# 
from sqlalchemy import Column, Integer, String
from .database import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)

# 
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str = None

class Item(BaseModel):
    id: int
    name: str
    description: str = None

    class Config:
        orm_mode = True

6. Service layer

In order to maintain the simplicity of the controller (routing processing function), business logic is separated into the service layer.

# services/item_service.py
from typing import List
from  import Session
from ..models import Item as ItemModel
from ..schemas import ItemCreate, Item

def get_items(db: Session, skip: int = 0, limit: int = 10) -> List[Item]:
    return (ItemModel).offset(skip).limit(limit).all()

def create_item(db: Session, item: ItemCreate) -> Item:
    db_item = ItemModel(**())
    (db_item)
    ()
    (db_item)
    return db_item

7. Routing grouping

Group relevant path operations into different router modules to improve code readability and maintainability.

# routers/
from fastapi import APIRouter, Depends
from  import Session
from ... import schemas, services
from ...dependencies import get_db

router = APIRouter()

@("/", response_model=List[])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    items = services.get_items(db, skip=skip, limit=limit)
    return items

@("/", response_model=)
def create_item(item: , db: Session = Depends(get_db)):
    return services.create_item(db=db, item=item)

8. Main application portal

Import and include individual routers in the main application file.

# 
from fastapi import FastAPI
from .routers import items, users

app = FastAPI()

app.include_router()
app.include_router()

9. Security

Implement authentication and authorization mechanisms, such as using JWT Token for user authentication.

# 
from datetime import datetime, timedelta
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from  import OAuth2PasswordBearer
from . import models, schemas, config

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = (token, .secret_key, algorithms=[])
        username: str = ("sub")
        if username is None:
            raise credentials_exception
        token_data = (username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

10. Test

Write unit tests and integration tests to make sure your API works as expected. FastAPI provides convenient testing clientsTestClient。

# tests/test_main.py
from  import TestClient
from my_project. import app

client = TestClient(app)

def test_read_main():
    response = ("/")
    assert response.status_code == 200
    assert () == {"message": "Hello World"}

11. Deployment

Consider containerizing your application with Docker and automatically deploying it to the cloud platform or server via a CI/CD pipeline.

# Dockerfile
FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9

COPY ./app /app
WORKDIR /app

RUN pip install --no-cache-dir -r 

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

Summarize

The above steps allow you to use FastAPI to build complex web APIs.

FastAPI supported tools

FastAPI provides a wealth of tools and features that make it easier to develop efficient, secure and easy to maintain Web APIs. In addition to core features, there are some important tools and features that can help you build more powerful applications. Here are some important tools and features of FastAPI:

1. Dependency injection system

FastAPI has built-in a powerful and flexible dependency injection system, which allows you to easily manage and reuse code logic such as authentication, database connections, etc.

from fastapi import Depends, FastAPI

app = FastAPI()

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        ()

@("/items/")
async def read_items(db: Session = Depends(get_db)):
    items = (Item).all()
    return items

2. Pydantic model

Pydantic is a data verification library that allows you to define data models and automatically process data parsing and verification. FastAPI uses Pydantic to ensure that request bodies, query parameters, and other inputs meet the expected format.

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

3. Middleware

Middleware is code executed before the request reaches the routing processing function or after the response is sent to the client. You can use middleware to add common behaviors such as logging, authentication, CORS processing, etc.

from  import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

4. Exception handling

FastAPI provides a built-in exception handling mechanism and allows you to customize exception handlers to handle specific types of errors.

from fastapi import Request, status
from  import JSONResponse
from  import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": (), "body": },
    )

5. Background task

If you need to continue performing certain operations (such as sending an email) after returning a response, you can use the background task.

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("", mode="a") as log:
        (message)

@("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}\n")
    return {"message": "Notification sent in the background"}

6. Static files and templates

FastAPI supports serving static files (such as HTML, CSS, JavaScript files) as well as rendering templates, which are useful for creating complete web applications.

from  import StaticFiles
from  import Jinja2Templates

("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")

@("/")
async def read_root(request: Request):
    return ("", {"request": request})

7. WebSocket Support

FastAPI supports WebSocket connections, which is useful for real-time two-way communications, such as chat applications or real-time updated data presentation.

from fastapi import WebSocket, WebSocketDisconnect

@("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await ()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        pass

8. Environment variables and configuration management

usepydanticofBaseSettingsClasses to manage application configuration and environment variables, which helps separate configuration and code.

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

    class Config:
        env_file = ".env"

settings = Settings()

9. Test tools

FastAPI provides convenient testing tools, including testing clientsTestClientand emulators to help you quickly test the API locally.

from  import TestClient

client = TestClient(app)

def test_read_main():
    response = ("/")
    assert response.status_code == 200
    assert () == {"message": "Hello World"}

10. OpenAPI and Swagger UI

FastAPI automatically generates OpenAPI specifications and provides an interactive API documentation interface (Swagger UI and ReDoc). This not only facilitates developers to debug the API, but also facilitates users to understand how to use your API.

# Automatically generate and provide interactive documents# Visit http://127.0.0.1:8000/doc or http://127.0.0.1:8000/redoc

11. Rate Limitation

While FastAPI itself does not directly provide rate limiting functionality, this feature can be implemented through middleware or other extensions to prevent abuse or protect server resources.

from fastapi import FastAPI, HTTPException, status
from  import Middleware
from  import TrustedHostMiddleware

app = FastAPI(middleware=[
    Middleware(TrustedHostMiddleware, allowed_hosts=[""])
])

# Or use third-party libraries such as fastapi-limiter to implement rate limiting

Together, these tools and features form a powerful and flexible ecosystem of FastAPI, allowing developers to build modern web APIs more efficiently.

FastAPI routing operations

FastAPI supports all standard HTTP methods (also known as "routing operations" or "HTTP verbs") that define the type of interaction between the client and the server. Here are the main routing operations supported by FastAPI:

1. GET

Used to request information from the server, usually used to query resources.

@("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

2. POST

Used to submit data to a specified resource, often used to create new resources or send form data.

@("/items/")
async def create_item(item: Item):
    return item

3. PUT

Used to update existing resources, usually the entire resource will be replaced.

@("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

4. DELETE

Used to delete the specified resource.

@("/items/{item_id}")
async def delete_item(item_id: int):
    return {"item_id": item_id}

5. PATCH

Used to partially update the resource and modify only the specified fields.

@("/items/{item_id}")
async def patch_item(item_id: int, item: ItemUpdate):
    return {"item_id": item_id, "item": item}

6. OPTIONS

Communication options used to describe the target resource. It returns a list of methods supported by the server.

@("/items/")
async def describe_options():
    return {"Allow": "GET, POST, PUT, DELETE, PATCH, OPTIONS"}

7. HEAD

Similar to a GET request, but does not return the message body and is only used to obtain response header information.

@("/items/{item_id}")
async def head_item(item_id: int):
    # Process logic, but will not return the response body    pass

8. TRACE

Used to echo the request received by the server, mainly for testing or diagnosis.

@("/trace")
async def trace_request():
    return {"method": "TRACE"}

Routing parameters

In addition to the above HTTP methods, FastAPI also allows you to define path parameters, query parameters, request bodies, etc. to handle different request scenarios more flexibly.

Path parameters

Define parameters directly in the path, such as/items/{item_id}Initem_id。

@("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Query parameters

Pass parameters through the query string of the URL, for example/items/?skip=0&limit=10。

@("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

Request body

Use the Pydantic model to parse and verify JSON request bodies.

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@("/items/")
async def create_item(item: Item):
    return item

Extra Features

  • Path operation function: Can be a synchronous function (def) or asynchronous function (async def), choose according to your needs.
  • Dependency injection: Dependencies can be added in the path operation function to facilitate sharing of state or performing preprocessing logic.
  • middleware: Add additional behavior to all requests, such as logging, authentication, etc.
  • Custom response: You can return different types of responses, including JSONResponse, HTMLResponse, etc.
  • WebSocket: FastAPI also supports WebSocket connections and can be used to implement real-time two-way communication.

The above are the main routing operations supported by FastAPI. FastAPI is designed to not only support standard HTTP methods, but also provide powerful tools to help developers build modern web APIs.

Summarize

This is the end of this article about how python builds complex web APIs through FastAPI. For more related content on python FastAPI, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!