Behavioral Pattern

Behavioral patterns focus on how objects interact and communicate with each other, ensuring that the system’s behavior is well-defined and predictable.

​

Observer Pattern

The Observer pattern defines a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.

Use Cases in FastAPI

• Event Handling: Triggering actions based on specific events.
• WebSocket Notifications: Broadcasting messages to connected clients.
• Background Tasks: Executing tasks in response to certain triggers.

Advantages

• Promotes loose coupling between subjects and observers.
• Enhances scalability by allowing dynamic addition/removal of observers.
• Facilitates broadcast communication.

Disadvantages

• Can lead to unexpected behavior if observers modify the subject.
• May become complex with many observers.

Observer Pattern with FastAPI: Event Handling and Notifications

Implementing a simple publish-subscribe mechanism for user registration.

from typing import List, Callable
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# Observer Interface
class Observer(ABC):
  @abstractmethod
  async def update(self, data: dict):
      pass

# Concrete Observers
class EmailNotifier(Observer):
  async def update(self, data: dict):
      print(f"Sending welcome email to {data['email']}")

class Logger(Observer):
  async def update(self, data: dict):
      print(f"User registered: {data['username']}")

# Subject
class UserSubject:
  def __init__(self):
      self._observers: List[Observer] = []

  def attach(self, observer: Observer):
      self._observers.append(observer)

  def detach(self, observer: Observer):
      self._observers.remove(observer)

  async def notify(self, data: dict):
      for observer in self._observers:
          await observer.update(data)

# Initialize Subject and Observers
user_subject = UserSubject()
user_subject.attach(EmailNotifier())
user_subject.attach(Logger())

# Pydantic Model
class User(BaseModel):
  username: str
  email: str

@app.post("/register/")
async def register_user(user: User):
  # Here, you would typically save the user to the database
  # After saving, notify observers
  await user_subject.notify(user.dict())
  return {"status": "User registered successfully"}

​

Strategy Pattern

The Strategy pattern defines a family of algorithms, encapsulates each one, and makes them interchangeable. This pattern allows the algorithm to vary independently from the clients that use it.

Use Cases in FastAPI

• Payment Processing: Switching between different payment gateways.
• Authentication Strategies: Supporting multiple authentication methods (e.g., JWT, OAuth).
• Data Validation: Applying different validation rules based on context.

Advantages

• Promotes algorithm encapsulation and interchangeability.
• Enhances code flexibility and reusability.
• Simplifies the addition of new strategies without modifying existing code.

Disadvantages

• Can lead to an increase in the number of classes.
• May complicate the design if strategies are not significantly different.

Strategy Pattern with FastAPI: Payment Processing

Implementing multiple payment strategies (PayPal and Stripe).

from abc import ABC, abstractmethod
from fastapi import FastAPI, HTTPException, Query

app = FastAPI()

# Abstract Strategy
class PaymentStrategy(ABC):
  @abstractmethod
  async def pay(self, amount: float) -> str:
      pass

# Concrete Strategies
class PayPalStrategy(PaymentStrategy):
  async def pay(self, amount: float) -> str:
      # Implement PayPal payment logic
      return f"Paid {amount} using PayPal"

class StripeStrategy(PaymentStrategy):
  async def pay(self, amount: float) -> str:
      # Implement Stripe payment logic
      return f"Paid {amount} using Stripe"

# Context
class PaymentContext:
  def __init__(self, strategy: PaymentStrategy):
      self._strategy = strategy

  async def execute_payment(self, amount: float) -> str:
      return await self._strategy.pay(amount)

# Strategy Factory
class PaymentStrategyFactory:
  @staticmethod
  def get_strategy(gateway: str) -> PaymentStrategy:
      if gateway == "paypal":
          return PayPalStrategy()
      elif gateway == "stripe":
          return StripeStrategy()
      else:
          raise ValueError("Unsupported payment gateway")

# FastAPI Route
@app.post("/pay/")
async def make_payment(amount: float, gateway: str = Query(...)):
  try:
      strategy = PaymentStrategyFactory.get_strategy(gateway)
      context = PaymentContext(strategy)
      result = await context.execute_payment(amount)
      return {"status": "success", "message": result}
  except ValueError as e:
      raise HTTPException(status_code=400, detail=str(e))

​

Note

This example does combine both the Strategy Pattern and the Factory Pattern effectively. Here’s how each pattern is used and how they work together:

Strategy Pattern

The Strategy Pattern is implemented through the PaymentStrategy interface and its concrete implementations (PayPalStrategy and StripeStrategy):

• PaymentStrategy: This is the abstract base class that defines a common pay method.

• Concrete Strategies (PayPalStrategy and StripeStrategy): Each of these classes provides a specific payment processing behavior, implementing the pay method with different logic for PayPal and Stripe payments.

• PaymentContext: This acts as the context class that accepts a PaymentStrategy instance and uses it to process the payment via execute_payment. This allows you to dynamically set or change the payment strategy.

Factory Pattern

The Factory Pattern is implemented in the PaymentStrategyFactory class:

• PaymentStrategyFactory: This class contains the static method get_strategy, which decides which PaymentStrategy to return based on the gateway parameter. This encapsulates the instantiation logic for different strategies, removing this responsibility from the main route handler.

• This factory method centralizes and abstracts the creation of the PaymentStrategy instances, allowing the client (in this case, the route) to request a strategy without knowing its specific details.

How They Work Together

• The FastAPI route handler (make_payment) accepts a gateway parameter, which determines the type of payment method.

• The PaymentStrategyFactory uses the gateway parameter to decide which PaymentStrategy to return, encapsulating the creation logic.

• The selected PaymentStrategy is then used within PaymentContext to perform the payment. The PaymentContext delegates the payment action to the strategy, allowing the behavior to vary based on the concrete strategy provided by the factory.