Leveraging the Singleton Design Pattern and Component Architecture in FastAPI Applications

Introduction

The Singleton pattern is a creational design pattern that ensures that only one instance of a class is created throughout the application. It provides a global point of access to this instance, making it useful for scenarios like managing a database connection or a shared resource.

Section 1: Understanding the Singleton Pattern

Definition and Code Example

class Singleton:
    _instance = None

    @staticmethod
    def get_instance():
        if Singleton._instance is None:
            Singleton._instance = Singleton()
        return Singleton._instance

Explanation

In the provided code example, we create a Singleton class with a private class attribute _instance that holds the reference to the single instance of the class. The get_instance() method is a static method that returns the instance of the class. It checks if _instance is None, and if so, it creates a new instance of the class using the class constructor. Subsequent calls to get_instance() return the existing instance.

Use Case

The Singleton pattern is commonly used in scenarios where you need to ensure that only one instance of a class exists throughout the application. For example, in a database application, you may want to have a single database connection shared among multiple components. By using the Singleton pattern, you can create a single instance of the database connection class, and all components can access this shared instance without the need to create multiple connections.

Section 2: Implementing Application Components using the Singleton Pattern

In this section, we will explore the various components that enhance the functionality and performance of our FastAPI application. These components include the Singleton pattern, an asyncio PostgreSQL database connection, a logger, a configuration manager, and a cache.

Asyncio PostgreSQL Database Connection

Code Example

When working with FastAPI and asyncio, utilizing an asyncio-based PostgreSQL database connection can greatly enhance the efficiency and concurrency of your database operations.

from databases import Database


class DatabaseManager:
    """
    DatabaseManager is a singleton class that manages the database connection.
    """
    _instance = None

    def __init__(self, database_url: str):
        self.database = Database(database_url, min_size=2, max_size=10)

    @staticmethod
    def get_instance(database_url: str):
        if DatabaseManager._instance is None:
            DatabaseManager._instance = DatabaseManager(database_url)
        return DatabaseManager._instance

    async def connect(self):
        await self.database.connect()

    async def disconnect(self):
        await self.database.disconnect()

Explanation: The DatabaseManager class encapsulates the logic for managing the asyncio PostgreSQL database connection. It takes the database_url as a parameter in the constructor and initializes the database  attribute. The connect method establishes the database connection by calling self.database.connect() and creates an asyncio connection pool . The disconnect method handles the disconnection from the database.

Use Case: The DatabaseManager class simplifies the management of the asyncio PostgreSQL database connection in your FastAPI application. By using a single instance of DatabaseManager throughout the application, you ensure efficient resource utilization and provide a centralized point of access for database operations. This promotes code reusability, better performance, and cleaner code organization.

Logger Manager

A logger is essential for recording and managing application logs. It allows you to track events, debug issues, and gather valuable information for analysis and troubleshooting.

Code Example

import logging


class LoggerManager:
    _instance = None

    def __init__(self, logger_name, file_name):
        self.logger = logging.getLogger(logger_name)
        self.logger.setLevel(logging.DEBUG)  # set logger level

        # add handlers
        self._add_console_handler()
        self._add_file_handler(file_name)

    @staticmethod
    def get_instance(logger_name, file_name):
        if LoggerManager._instance is None:
            LoggerManager._instance = LoggerManager(logger_name, file_name)
        return LoggerManager._instance

    def _add_console_handler(self):
        # create console handler and set level to debug
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)

        # create and add formatter to ch
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        ch.setFormatter(formatter)

        # add ch to logger
        self.logger.addHandler(ch)

    def _add_file_handler(self, file_name):
        # create file handler which logs even debug messages
        fh = logging.FileHandler(file_name)
        fh.setLevel(logging.DEBUG)

        # create and add formatter to fh
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        fh.setFormatter(formatter)

        # add fh to logger
        self.logger.addHandler(fh)

Explanation: The LoggerManager class provides a simple way to manage the logger throughout the application. It initializes the logger instance in the constructor, using the provided logger_name and file_name. You can configure the logger further based on your specific logging requirements, such as setting the log level or adding handlers.

Use Case: By utilizing the LoggerManager class, you centralize the management of the logger instance, allowing consistent logging behavior across different parts of your application. This promotes better code organization, simplified debugging, and streamlined log management. You can easily access the logger instance wherever needed and utilize its methods to log relevant information, warnings, or errors.

Configuration Manager

Managing configuration settings in a structured manner is crucial for an application's flexibility and maintainability. The ConfigurationManager class provides a centralized way to handle configuration settings in your FastAPI application.

Code Example

from pydantic_settings import BaseSettings


class ConfigurationManager(BaseSettings):
    title: str
    description: str
    version: str

    class Config:
        env_file = ".env.fastapi"


_instance = None


def get_configuration():
    global _instance
    if _instance is None:
        _instance = ConfigurationManager()
    return _instance

Explanation: The ConfigurationManager class extends the BaseSettings class from the pydantic library to define and manage the configuration settings. It includes the required app_name and app_version attributes. You can add additional configuration settings as per your application's needs.

Use Case: By using the ConfigurationManager class, you centralize the management of configuration settings, making it easier to access and utilize these settings across different parts of your application. It promotes code reusability, consistent configuration handling, and provides a single point of control for managing the application's settings.

Cache Manager

Caching frequently accessed data can greatly enhance the performance of your API endpoints by reducing response time and server load.

Code Example

class CacheManager:
    _instance = None

    def __init__(self):
        self.cache = {}

    @staticmethod
    def get_instance():
        if CacheManager._instance is None:
            CacheManager._instance = CacheManager()
        return CacheManager._instance

    def get(self, key):
        return self.cache.get(key)

    def set(self, key, value):
        self.cache[key] = value

Explanation: The CacheManager class provides a simple caching mechanism by utilizing an in-memory cache, represented by the self.cache dictionary. It includes the get and set methods to retrieve and store values in the cache, respectively.

Use Case: By incorporating the CacheManager class, you can cache frequently accessed data in your API endpoints, reducing the need for repetitive computations or expensive operations. This results in faster response times, reduced server load, and improved overall performance of your application.

After separating the classes into separate files, you can import them into main.py as follows:

from fastapi import FastAPI
from database_manager import DatabaseManager
from logger_manager import LoggerManager
from config_manager import ConfigurationManager
from cache_manager import CacheManager

app = FastAPI()

# Create instances of the managers
database_manager = DatabaseManager(database_url="postgresql://user:password@localhost/db_name")
logger_manager = LoggerManager(logger_name="fastapi_app")
config_manager = ConfigurationManager(app_name="MyApp", app_version="1.0.0")
cache_manager = CacheManager()

# Define startup and shutdown event functions
@app.on_event("startup")
async def startup():
    await database_manager.connect()

@app.on_event("shutdown")
async def shutdown():
    await database_manager.disconnect()

# Example route that retrieves data from a table using the database connection
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = "SELECT * FROM users WHERE id = $1"
    async with database_manager.pool.acquire() as connection:
        result = await connection.fetchrow(query, user_id)
    return result

# Example route that uses the logger
@app.get("/logs")
def get_logs():
    logger = logger_manager.logger
    logger.info("Fetching logs from the API")
    return "Logs"

# Example route that retrieves the configuration
@app.get("/config")
def get_config():
    app_name = config_manager.app_name
    app_version = config_manager.app_version
    return {"app_name": app_name, "app_version": app_version}

# Examples route that interacts with the cache
@app.get("/trend-data")
def get_trend_data():
    cached_data = cache_manager.get("trend_data")

    if cached_data:
        return cached_data
    else:
        # Retrieve the trend data from your data source
        trend_data = fetch_trend_data()

        # Cache the trend data with an expiration time of one month
        cache_manager.set("trend_data", trend_data, expiration=2592000)  # Expiration time in seconds (30 days)

        return trend_data
        
@app.get("/cache/{key}")
def get_cache(key: str):
    value = cache_manager.get(key)
    if value is not None:
        return value
    else:
        # Perform the expensive operation to compute the value
        value = perform_expensive_operation()

        # Store the value in the cache
        cache_manager.set(key, value)

        return value

Bonus Section: Redis Integration for Cache Management

In this bonus section, we will enhance the CacheManager class to support using a Redis database as the caching storage. Redis is an in-memory data structure store that can be used as a highly efficient and scalable cache solution.

Code Example

import redis

class CacheManager:
    _instance = None

    def __init__(self, use_redis=False):
        self.cache = {}
        self.redis_client = None
        self.use_redis = use_redis

        if self.use_redis:
            self.redis_client = redis.Redis()

    @staticmethod
    def get_instance(use_redis=False):
        if CacheManager._instance is None:
            CacheManager._instance = CacheManager(use_redis=use_redis)
        return CacheManager._instance

    def get(self, key):
        if self.use_redis:
            cached_data = self.redis_client.get(key)
            if cached_data is not None:
                return cached_data.decode()

        return self.cache.get(key)

    def set(self, key, value):
        if self.use_redis:
            self.redis_client.set(key, value)

        self.cache[key] = value

Explanation: The updated CacheManager class now includes an optional integration with Redis for caching purposes. The use_redis parameter is added to the constructor to enable or disable the Redis integration. If use_redis is set to True, the class creates a Redis client instance using redis.Redis(). The get method is modified to first check if the cache data exists in Redis using self.redis_client.get(key). If the data is found, it is returned after decoding it. The set method is updated to store the data in Redis using self.redis_client.set(key, value), if the Redis integration is enabled.

Use Case: By incorporating the Redis integration option in the CacheManager class, you can leverage the capabilities of Redis for efficient caching. Redis provides advanced features like data expiration, efficient data structures, and distributed caching options, which can significantly enhance the performance and scalability of your caching solution.

To use the Redis integration, instantiate the CacheManager with use_redis=True:

cache_manager = CacheManager.get_instance(use_redis=True)

With Redis as part of the cache management, you can take advantage of its fast and scalable in-memory storage to handle caching needs efficiently. Redis integration allows you to store larger amounts of data in the cache, provide expiration settings, and leverage Redis's rich data structures and features.

Note: To use Redis with Python, make sure to have the redis library installed (pip install redis). Additionally, you may need to provide the Redis server connection details in the Redis client initialization if it's not running on the default localhost.

By incorporating Redis as an option in the CacheManager class, you have the flexibility to choose between an in-memory cache or a Redis-based cache based on your specific requirements and the scalability needs of your application.

Here is a docker-compose file you can uses as part of your testing.

version: '3.8'

services:
  postgresql:
    image: postgres:14.2-alpine
    hostname: test_db
    shm_size: 2g
    restart: unless-stopped
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: ${POSTGRES_DB}
    volumes:
      - "./postgres-data:/var/lib/postgresql/data"
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U postgres" ]
      interval: 30s
      timeout: 30s
      retries: 5
  redis:
    image: redis:6.0.20-alpine
    hostname: test_redis
    restart: unless-stopped
    ports:
      - "6379:6379"
    volumes:
      - "./redis-data:/data"

Conclusion

Incorporating these application components within your FastAPI application enhances its functionality, performance, and maintainability. The Singleton pattern ensures the uniqueness and centralized access of important resources, such as the database connection, logger, configuration settings, and cache. By utilizing these components, you achieve better code organization, reduced duplication, and improved control over critical aspects of your application.

Subscribe to Devtooler

Don’t miss out on the latest issues. Sign up now to get access to the library of members-only issues.
[email protected]
Subscribe