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.