Build a solid understanding of FastAPI

FastAPI is a modern, fast web framework for building APIs with Python 3.7+ based on standard Python type hints. The key features of FastAPI include:

• Fast to run: High performance, on par with NodeJS and Go (thanks to Starlette and Pydantic). One of the fastest Python frameworks available.
• Fast to code: Great editor support. Completion everywhere. Less time debugging.
• Robust: Get production-ready code with automatic interactive documentation.
• Standards-based: Based on the open standards for APIs: OpenAPI (previously known as Swagger) and JSON Schema.

1. Python Types Intro

FastAPI is built on Python 3.7+ type hints, enabling automatic request validation, serialization, and documentation generation. Type hints help developers ensure that their codebases are more maintainable and error-resistant. Pydantic uses these type hints to perform data validation and serialization, converting input data (like JSON) to Python types and vice versa.

from pydantic import BaseModel

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

@app.post("/items/")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}        

In this example, Item is a Pydantic model with type annotations. FastAPI uses these annotations to validate incoming data and serialize outgoing data.

2. Concurrency and async/await

Concurrency in FastAPI is handled through Python’s asyncio library. The async and await syntax allows you to write non-blocking code that can perform multiple operations at the same time. This is particularly useful for I/O-bound tasks such as database operations, file handling, or network requests, improving the performance of your API by not wasting time waiting for these operations to complete.

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/data/")
async def fetch_data():
    async with httpx.AsyncClient() as client:
        result = await client.get("https://api.example.com/data")
        return result.json()        

This endpoint uses async and await to fetch data from a remote API asynchronously, ensuring the server can handle other requests while waiting for the response.

3. Dependencies

Dependencies in FastAPI are reusable components that can be injected into your path operation functions. This system allows for cleaner code and easier testing. Dependencies can be anything from database sessions and API clients to complex authentication systems. They are defined as functions that return a value, which gets passed to your endpoint as an argument.

from fastapi import FastAPI, Depends, HTTPException

def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons        

Here, common_parameters is a dependency that simplifies the management of common query parameters.

4. Security

FastAPI provides several tools to implement security in your applications, from basic authentication to OAuth2. FastAPI simplifies integrating these security schemes with your API, ensuring secure access and operation. It provides extensive support for security tokens, including JWT tokens and OAuth2 with password and bearer with JWT tokens.

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def token(form_data: OAuth2PasswordRequestForm = Depends()):
    return {"access_token": form_data.username, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}        

This code snippet demonstrates how to implement a simple OAuth2 authentication system with token generation and protected endpoints.

5. Middleware

Middleware are functions that run before every request is processed. They can manipulate the request before it reaches your endpoint or manipulate the response before it is sent back to the client. Examples include managing sessions, adding headers to responses, or logging requests for monitoring.

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

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

This middleware setup allows cross-origin requests from any domain, crucial for API accessibility from different frontends.

6. CORS (Cross-Origin Resource Sharing)

CORS is a security feature that restricts how resources on a web page can be requested from another domain outside the domain from which the first resource was served. FastAPI provides a middleware to manage CORS, allowing you to specify which domains can access your API, which methods they can use, and whether to allow credentials.

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

# List of allowed origins (i.e., the URLs that can access the API)
origins = [
    "https://localhost:3000",  # Allow local development frontend access
    "https://www.example.com",
]

# Adding CORS middleware to FastAPI application
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,  # List of origins that can access the API
    allow_credentials=True,  # Whether to allow credentials (cookies, authorization headers, etc.)
    allow_methods=["*"],  # Which methods can be used to access the API, ["*"] allows all
    allow_headers=["*"],  # Which headers can be used when making a request
)

@app.get("/data/")
async def get_data():
    return {"message": "This is a CORS-enabled response!"}        

Key Points in the Example:

1. Origins: The origins list defines which websites are allowed to make requests to your API. In the example, a local development server (`https://localhost:3000`) and a production domain (`https://www.example.com`) are allowed.

2. Middleware Configuration:

- allow_origins: Specifies which domains are allowed to access the API.

- allow_credentials: If set to True, it allows cookies, authorization headers, or TLS client certificates to be sent with the requests.

- allow_methods: Specifies which HTTP methods are allowed. Using ["*"] means all methods (GET, POST, etc.) are allowed.

- allow_headers: Specifies which headers are allowed in requests. Again, using ["*"] means all headers are permitted.

This setup ensures that your FastAPI application can handle requests from specified domains, with specific configurations about what those requests can include and how they can interact with your API. This is crucial for ensuring that your API can be securely accessed from web-based clients hosted on different origins.

7. SQL (Relational) Databases

FastAPI doesn't include a specific database in its framework, but it integrates seamlessly with any SQL database by using databases like SQLAlchemy. You can use SQLAlchemy for ORM (Object-Relational Mapping), which abstracts the database interactions into Python code, making database operations easier to manage and more secure.

Define Database Models and Schema

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    items = relationship("Item", back_populates="owner")

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")        

Create Database Tables

Base.metadata.create_all(bind=engine)        

CRUD Operations

from sqlalchemy.orm import Session

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def create_user_item(db: Session, item: Item, user_id: int):
    db_item = Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item        

Use Models in FastAPI Endpoints

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/{user_id}/items/", response_model=Item)
def create_item_for_user(user_id: int, item: Item, db: Session = Depends(get_db)):
    return create_user_item(db=db, item=item, user_id=user_id)        

Key Components

• SQLAlchemy ORM: Allows you to interact with your database using Python code instead of SQL.
• Session Management: Manages database sessions through dependencies in FastAPI.
• CRUD Functions: Facilitate creating, reading, updating, and deleting database entries.

8. Background Tasks

Background tasks in FastAPI let you handle operations after a response has been sent to the client. This is useful for operations that need to happen after a request but whose timing does not affect the response to the client, such as sending emails or processing data.

def write_log(message: str):
    with open("log.txt", "a") as file:
        file.write(f"{message}\n")        

This simple function just writes a message to a file, simulating a logging operation.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

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

In this example:

• The /send-notification/ endpoint takes an email address as input.
• It uses BackgroundTasks to add the write_log function to the tasks that should be executed after the response is sent.
• The write_log function is called with the message argument detailing the notification process.

Key Points to Consider

• Non-blocking: Background tasks allow your endpoint to respond immediately, improving the perceived performance of your API.
• Resource Management: Make sure that the tasks you run in the background do not exhaust your server resources, as they still run in the same server process.
• Error Handling: Errors in background tasks won’t affect the response but should be properly handled or logged to ensure they do not pass unnoticed.

Background tasks are a powerful feature in FastAPI for improving API responsiveness and efficiently managing long-running operations.

9. Testing

Testing in FastAPI is facilitated by Starlette's test client. This allows you to send test requests to your API, receive responses, and inspect the results in your tests. This makes unit testing and integration testing much simpler and can be integrated into your CI/CD pipeline.

from fastapi.testclient import TestClient
from .main import app  # Import your FastAPI application here

client = TestClient(app)

def test_send_notification():
    response = client.post("/send-notification/", json={"email": "[email protected]"})
    assert response.status_code == 200
    assert response.json() == {"message": "Notification being processed in the background"}        

def test_create_item():
    response = client.post("/items/", json={"name": "item1", "description": "A test item"})
    assert response.status_code == 200
    assert response.json()["name"] == "item1"

def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json()["name"] == "item1"

def test_update_item():
    response = client.put("/items/1", json={"name": "updated item1", "description": "Updated description"})
    assert response.status_code == 200
    assert response.json()["name"] == "updated item1"

def test_delete_item():
    response = client.delete("/items/1")
    assert response.status_code == 200
    assert response.json() == {"message": "Item deleted"}        

10. Advanced Dependencies

Advanced dependencies in FastAPI allow for more complex dependency injection scenarios. These include dependencies with sub-dependencies, dependencies that require "cleanup" after the request, and dependencies that are only applied under certain conditions.

11. WebSockets

WebSockets provide a way to open a bi-directional, persistent connection between the client and server. FastAPI supports WebSockets natively, allowing you to handle WebSocket sessions and manage data transmission in real-time, which is perfect for applications like chat systems or live notifications.

12. Including WSGI - Flask, Django, others

FastAPI can run alongside WSGI applications like Flask and Django, allowing you to combine the synchronous and asynchronous worlds. This is achieved using an ASGI-to-WSGI adapter that lets you mount WSGI apps as sub-applications within an ASGI application like FastAPI.

13. Run a Server Manually - Uvicorn

Uvicorn is an ASGI server that is recommended for running FastAPI applications. It's lightweight, super-fast, and supports ASGI out of the box. Running a FastAPI app manually with Uvicorn typically involves calling Uvicorn with your application instance.

14. Server Workers - Gunicorn with Uvicorn

For production deployments, Gunicorn 'Green Unicorn' is used as a WSGI HTTP server to manage Uvicorn and handle multiple worker processes. Gunicorn with Uvicorn allows you to scale your application by running multiple Uvicorn workers, handling more requests simultaneously.

These components and concepts illustrate how FastAPI combines modern Python features and standards to provide a powerful toolkit for building APIs. Each part is designed to provide performance, ease of use, and flexibility in your development process.