Creational patterns deal with object creation mechanisms, trying to create objects in a manner suitable to the situation. They abstract the instantiation process, making a system independent of how its objects are created.

​
Singleton Pattern

  • The Singleton pattern ensures that a class has only one instance and provides a global point of access to it.
  • Use Cases:
    • Database Connections: Managing a single database connection pool.
    • Configuration Management: Ensuring a single configuration instance is used throughout the application.

Basic Singleton in Python

class Singleton:
  _instance = None

  def __new__(cls, *args, **kwargs):
      if not cls._instance:
          cls._instance = super(Singleton, cls).__new__(cls)
      return cls._instance

# Usage
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # Output: True

Singleton with Thread Safety For multi-threaded applications, ensure thread safety to prevent multiple instances.

import threading

class Singleton:
  _instance = None
  _lock = threading.Lock()

  def __new__(cls, *args, **kwargs):
      with cls._lock:
          if not cls._instance:
              cls._instance = super(Singleton, cls).__new__(cls)
      return cls._instance

# Usage remains the same

Practical Example in FastAPI: Database Connection Management

from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker

class Database:
  _engine: AsyncEngine = None
  _SessionLocal = None

  @classmethod
  def initialize(cls, DATABASE_URL: str):
      if not cls._engine:
          cls._engine = create_async_engine(DATABASE_URL, echo=True)
          cls._SessionLocal = sessionmaker(
              bind=cls._engine,
              class_=AsyncSession,
              expire_on_commit=False
          )

  @classmethod
  def get_engine(cls):
      if not cls._engine:
          raise Exception("Database not initialized.")
      return cls._engine

  @classmethod
  def get_session(cls):
      if not cls._SessionLocal:
          raise Exception("Database not initialized.")
      return cls._SessionLocal()

from fastapi import FastAPI, Depends

app = FastAPI()

# Initialize Database at startup
@app.on_event("startup")
async def startup():
  Database.initialize("postgresql+asyncpg://user:password@localhost/dbname")

# Dependency
async def get_db():
  async with Database.get_session() as session:
      yield session

@app.get("/users/")
async def get_users(db: AsyncSession = Depends(get_db)):
  result = await db.execute(select(UserModel))
  users = result.scalars().all()
  return users

​
Factory Pattern

The Factory pattern defines an interface for creating objects but allows subclasses to alter the type of objects that will be created. It promotes loose coupling by eliminating the need to bind application-specific classes into the code.

Use Cases in FastAPI

  • Creating Service Classes: Depending on the context, instantiate different service implementations.
  • Database Model Factories: Dynamically creating different database models based on configuration.
  • API Response Factories: Generating responses based on various conditions or formats.

Advantages

  • Encapsulates object creation.
  • Promotes code reusability and flexibility.
  • Facilitates easy addition of new types without modifying existing code.

Disadvantages

  • Can introduce complexity if overused.
  • Might obscure the code flow, making it harder to trace object creations.

Factory Pattern with FastAPI: Service Creation

from abc import ABC, abstractmethod

# Abstract Base Class
class PaymentGateway(ABC):
  @abstractmethod
  async def process_payment(self, amount: float):
      pass

# Concrete Implementations
class PayPalGateway(PaymentGateway):
  async def process_payment(self, amount: float):
      # Implement PayPal payment processing
      return f"Processed {amount} via PayPal"

class StripeGateway(PaymentGateway):
  async def process_payment(self, amount: float):
      # Implement Stripe payment processing
      return f"Processed {amount} via Stripe"

# Factory
class PaymentGatewayFactory:
  @staticmethod
  def get_gateway(gateway_name: str) -> PaymentGateway:
      if gateway_name == "paypal":
          return PayPalGateway()
      elif gateway_name == "stripe":
          return StripeGateway()
      else:
          raise ValueError("Unsupported payment gateway")

# FastAPI Route
from fastapi import FastAPI, HTTPException, Query

app = FastAPI()

@app.post("/pay/")
async def make_payment(amount: float, gateway: str = Query(...)):
  try:
      payment_gateway = PaymentGatewayFactory.get_gateway(gateway)
      result = await payment_gateway.process_payment(amount)
      return {"status": "success", "message": result}
  except ValueError as e:
      raise HTTPException(status_code=400, detail=str(e))

​
Builder Pattern

The Builder pattern separates the construction of a complex object from its representation, allowing the same construction process to create different representations.

Use Cases in FastAPI

  • Complex Object Construction: Building objects with numerous optional parameters.
  • Request/Response Builders: Dynamically assembling responses based on various conditions.
  • Query Builders: Creating complex database queries programmatically.

Advantages

  • Provides precise control over the construction process.
  • Simplifies the creation of complex objects.
  • Enhances code readability and maintainability.

Disadvantages

  • Can introduce additional classes and complexity.
  • Might be overkill for simple object creation scenarios.

Builder Pattern with FastAPI: Request Filtering Imagine an API that allows users to filter items based on various optional parameters.

from typing import Optional
from fastapi import FastAPI, Query
from sqlalchemy.orm import Session

app = FastAPI()

# Assume ItemModel is a SQLAlchemy model

class QueryBuilder:
  def __init__(self, base_query):
      self.query = base_query

  def filter_by_name(self, name: Optional[str]):
      if name:
          self.query = self.query.filter(ItemModel.name == name)
      return self

  def filter_by_price_range(self, min_price: Optional[float], max_price: Optional[float]):
      if min_price is not None:
          self.query = self.query.filter(ItemModel.price >= min_price)
      if max_price is not None:
          self.query = self.query.filter(ItemModel.price <= max_price)
      return self

  def order_by_price(self, ascending: bool = True):
      if ascending:
          self.query = self.query.order_by(ItemModel.price)
      else:
          self.query = self.query.order_by(ItemModel.price.desc())
      return self

  def build(self):
      return self.query

@app.get("/items/")
async def get_items(
  name: Optional[str] = Query(None),
  min_price: Optional[float] = Query(None, ge=0),
  max_price: Optional[float] = Query(None, ge=0),
  sort_ascending: bool = True,
  db: Session = Depends(get_db)
):
  builder = QueryBuilder(db.query(ItemModel))
  items_query = (
      builder
      .filter_by_name(name)
      .filter_by_price_range(min_price, max_price)
      .order_by_price(sort_ascending)
      .build()
  )
  items = items_query.all()
  return items

Was this page helpful?