Optimizing FastAPI Applications: Modular Design, Logging, and Testing

Avatar

By squashlabs, Last Updated: June 21, 2023

Optimizing FastAPI Applications: Modular Design, Logging, and Testing

Introduction:
FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints. It is designed to be easy to use and highly efficient, making it a popular choice for developing web applications. In this article, we will explore some best practices for optimizing FastAPI applications, including modular design, logging, and testing.

Dependency Injection in FastAPI:
Dependency Injection (DI) is a design pattern that allows the separation of the creation of an object from its dependencies. FastAPI provides built-in support for DI using the Depends class from the fastapi module. By using DI, we can easily inject dependencies into our FastAPI application, making it more modular and easier to test.

To demonstrate DI in FastAPI, let’s consider a simple example where we have a service class that requires a database connection. We can define a function that creates the database connection and use the Depends class to inject it into our service class.

from fastapi import Depends, FastAPI

app = FastAPI()

def get_database_connection():
    # Create and return the database connection
    connection = create_database_connection()
    return connection

class MyService:
    def __init__(self, db: Database = Depends(get_database_connection)):
        self.db = db

@app.get("/")
async def index(service: MyService = Depends()):
    # Use the service with the injected database connection
    result = service.do_something()
    return {"result": result}

In the example above, we define a function get_database_connection that creates a database connection. We then define our MyService class that requires a db parameter of type Database. We use the Depends class to inject the database connection into our service class. Finally, in our route handler function, we define a parameter service of type MyService and FastAPI will automatically inject the service instance with the database connection.

Authentication and Authorization in FastAPI:
Authentication and authorization are crucial aspects of any web application. FastAPI provides built-in support for implementing authentication and authorization using various authentication methods such as API keys, OAuth2, JWT, etc.

To demonstrate authentication and authorization in FastAPI, let’s consider an example where we have an endpoint that requires authentication using JWT tokens. We can use the Depends class to inject a function that validates the JWT token and extracts the user information.

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, secret_key, algorithms=[algorithm])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid authentication credentials",
                    headers={"WWW-Authenticate": "Bearer"},
                )
        return username
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    # Access the protected resource using the authenticated user
    return {"message": f"Hello {current_user}"}

In the example above, we define a function get_current_user that takes a JWT token as a parameter and validates it using the jwt.decode function. If the token is valid, we extract the username from the payload and return it. If the token is invalid, we raise an HTTPException with a status code of 401 Unauthorized. In our route handler function, we define a parameter current_user of type str and FastAPI will automatically inject the authenticated username.

Caching Strategies in FastAPI:
Caching is an important technique to improve the performance and scalability of web applications. FastAPI provides support for implementing caching strategies using various caching backends such as Redis, Memcached, etc.

To demonstrate caching in FastAPI, let’s consider an example where we have an endpoint that retrieves data from a slow database query. We can use the Depends class to inject a caching function that checks if the data is already cached and returns it, otherwise, it retrieves the data from the database and stores it in the cache.

from fastapi import Depends, FastAPI
from fastapi_cache import FastAPICache, caches, close_caches
from fastapi_cache.backends.redis import RedisBackend

app = FastAPI()
cache = FastAPICache()

redis_cache = RedisBackend("redis://localhost:6379/0")
caches.set("default", redis_cache)

def get_data_from_database():
    # Simulate slow database query
    import time
    time.sleep(5)
    return {"data": "Some data"}

def get_data_from_cache_or_database():
    data = cache.get("data")
    if data:
        return data
    else:
        data = get_data_from_database()
        cache.set("data", data, expire=60)
        return data

@app.get("/")
async def index(data: dict = Depends(get_data_from_cache_or_database)):
    # Use the data retrieved from the cache or database
    return {"data": data}

In the example above, we define a function get_data_from_database that simulates a slow database query. We then define a function get_data_from_cache_or_database that checks if the data is already cached using the cache.get function. If the data is found in the cache, it is returned, otherwise, it retrieves the data from the database, stores it in the cache using the cache.set function, and returns it. In our route handler function, we define a parameter data of type dict and FastAPI will automatically inject the data retrieved from the cache or database.

Performance Optimization Techniques in FastAPI:
FastAPI is designed to be highly efficient and performant out of the box. However, there are some performance optimization techniques that can further improve the performance of FastAPI applications.

1. Use asynchronous code: FastAPI supports asynchronous code using Python’s asyncio library. By using async and await keywords, you can write non-blocking code that allows your application to handle more requests concurrently.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def index():
    # Perform asynchronous operations
    result = await perform_async_operation()
    return {"result": result}

In the example above, we define an asynchronous route handler function using the async keyword. Inside the function, we use the await keyword to perform asynchronous operations. This allows other requests to be processed while waiting for the result of the asynchronous operation.

2. Use connection pooling: FastAPI supports connection pooling for database connections using libraries like SQLAlchemy. Connection pooling allows you to reuse existing connections instead of creating new ones for each request, reducing the overhead of establishing a new connection.

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()

database_url = "sqlite:///./mydatabase.db"
engine = create_engine(database_url, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

@app.get("/")
async def index(db: Session = Depends(get_db)):
    # Use the database connection
    result = db.query(...)
    return {"result": result}

In the example above, we define a function get_db that creates a database session using SQLAlchemy’s sessionmaker class. We use the yield keyword to create a generator function that provides the database session to the route handler function. FastAPI will automatically inject the database session into the db parameter of the route handler function.

Database Connection Pooling in FastAPI:
FastAPI supports connection pooling for database connections using libraries like SQLAlchemy. Connection pooling allows you to reuse existing connections instead of creating new ones for each request, reducing the overhead of establishing a new connection.

To demonstrate database connection pooling in FastAPI, let’s consider an example where we have a route that queries a database using SQLAlchemy.

from fastapi import Depends, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

app = FastAPI()

database_url = "sqlite:///./mydatabase.db"
engine = create_engine(database_url, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

@app.get("/users")
async def get_users(db: Session = Depends(get_db)):
    # Query the database for users
    users = db.query(User).all()
    return {"users": users}

In the example above, we define a function get_db that creates a database session using SQLAlchemy’s sessionmaker class. We use the yield keyword to create a generator function that provides the database session to the route handler function. FastAPI will automatically inject the database session into the db parameter of the route handler function. The pool_pre_ping=True argument enables connection pooling with automatic pinging to check if the connection is still valid before using it.

Rate Limiting in FastAPI:
Rate limiting is an important technique to control the number of requests a client can make to a web application within a certain time window. FastAPI provides built-in support for rate limiting requests using the RateLimiter class.

To demonstrate rate limiting in FastAPI, let’s consider an example where we have an endpoint that is rate-limited to 100 requests per minute.

from fastapi import Depends, FastAPI, HTTPException
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

app = FastAPI()
limiter = FastAPILimiter(app)

@app.get("/")
@limiter.limit("100/minute")
async def index():
    # Handle the request
    return {"message": "Hello, world!"}

In the example above, we define a route handler function for the root endpoint (“/”) and use the @limiter.limit decorator to apply rate limiting. The argument to the limit decorator specifies the rate limit in the format “requests/window”, where “requests” is the maximum number of requests allowed and “window” is the time window in which the requests are counted. If the rate limit is exceeded, a HTTPException with a status code of 429 Too Many Requests is raised.

Handling File Uploads in FastAPI:
FastAPI provides built-in support for handling file uploads. You can define a route that accepts file uploads using the UploadFile class from the fastapi module.

To demonstrate handling file uploads in FastAPI, let’s consider an example where we have an endpoint that accepts an image file upload and saves it to the server.

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # Save the uploaded file
    contents = await file.read()
    with open(file.filename, "wb") as f:
        f.write(contents)
    return {"filename": file.filename}

In the example above, we define a route handler function that accepts a file upload parameter file of type UploadFile. The UploadFile class provides methods to read the contents of the uploaded file. We use the await file.read() function to asynchronously read the contents of the file. We then open a file with the same filename as the uploaded file and write the contents to it. Finally, we return a JSON response with the filename of the uploaded file.

Background Tasks and Scheduling in FastAPI:
FastAPI provides support for running background tasks and scheduling tasks to be executed at specific times using the BackgroundTasks class.

To demonstrate background tasks and scheduling in FastAPI, let’s consider an example where we have an endpoint that triggers a background task to send an email.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_email(email: str, message: str):
    # Send the email
    ...

@app.post("/send-email")
async def send_email_route(email: str, message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, message)
    return {"message": "Email sent in the background"}

In the example above, we define a function send_email that sends an email. We then define a route handler function that accepts an email and a message as parameters. We use the background_tasks parameter of type BackgroundTasks to add the send_email function as a background task using the add_task method. The background task will be executed asynchronously in the background while the route handler function returns a response to the client.

API Versioning in FastAPI:
API versioning is an important aspect of API development to ensure backward compatibility and smooth transitions between different versions of an API. FastAPI provides built-in support for API versioning using path parameters.

To demonstrate API versioning in FastAPI, let’s consider an example where we have two versions of an API, v1 and v2.

from fastapi import FastAPI

app = FastAPI()

@app.get("/v1/items/{item_id}")
async def get_item_v1(item_id: int):
    # Get item from v1 API
    ...

@app.get("/v2/items/{item_id}")
async def get_item_v2(item_id: int):
    # Get item from v2 API
    ...

In the example above, we define two route handler functions, one for each version of the API. The path parameter {item_id} is used to identify the item to retrieve. Clients can access the different versions of the API by specifying the version in the URL, for example, /v1/items/1 or /v2/items/1.

Error Monitoring and Alerting in FastAPI:
Error monitoring and alerting are critical for maintaining the stability and reliability of a web application. FastAPI provides support for error monitoring and alerting using popular error monitoring services such as Sentry, New Relic, etc.

To demonstrate error monitoring and alerting in FastAPI, let’s consider an example where we have an endpoint that raises an exception and sends an error report to an error monitoring service.

from fastapi import FastAPI
import sentry_sdk
from sentry_sdk.integrations.asgi import SentryAsgiMiddleware

app = FastAPI()

sentry_sdk.init(dsn="YOUR-SENTRY-DSN")

@app.get("/")
async def index():
    # Simulate an exception
    raise Exception("Something went wrong")

In the example above, we initialize the Sentry SDK with our Sentry DSN (Data Source Name). We then define a route handler function that raises an exception. The exception is automatically captured by the Sentry SDK and sent to the configured Sentry project for error monitoring and alerting.

Additional Resources

Dependency Injection in FastAPI
Authentication and Authorization in FastAPI
Database Migrations in FastAPI

More Articles from the Python Tutorial: From Basics to Advanced Concepts series:

How to Use Matplotlib for Chinese Text in Python

This guide provides a concise overview of using Matplotlib to render Chinese text in Python. It covers essential topics, including setting up Chinese fonts, configuring... read more

How To Exit/Deactivate a Python Virtualenv

Learn how to exit a Python virtualenv easily using two simple methods. Discover why you might need to exit a virtual environment and explore alternative ideas and best... read more

How to Integrate Python with MySQL for Database Queries

Pair Python with MySQL to execute database queries effortlessly. Learn about the Python MySQL Connector, establishing connections, interacting with MySQL, and more.... read more

16 Amazing Python Libraries You Can Use Now

In this article, we will introduce you to 16 amazing Python libraries that are widely used by top software teams. These libraries are powerful tools that can enhance... read more

Database Query Optimization in Django: Boosting Performance for Your Web Apps

Optimizing database queries in Django is essential for boosting the performance of your web applications. This article explores best practices and strategies for... read more

Converting Integer Scalar Arrays To Scalar Index In Python

Convert integer scalar arrays to scalar index in Python to avoid the 'TypeError: Only integer scalar arrays can be converted to a scalar index with 1D' error. This... read more