Abstract Factory Example: Storage and Queues per Cloud Provider

Switch from one provider to another by changing the one factory the application is given.

An application that runs on either AWS or Google Cloud is a natural Abstract Factory, and this example maps the template onto it. Archiving an order stores its receipt in object storage and publishes a message on a queue, and both must come from the same provider. The provider is the family, and storage and queue are the two kinds of product.

Template participants and the classes that play them
Participant	Class in this example
AbstractFactory	`CloudFactory`, with `create_storage` and `create_queue`
ConcreteFactory	`AwsFactory` and `GcpFactory`
AbstractProduct	`Storage` and `Queue`
ConcreteProduct	`S3Storage` and `SqsQueue` for AWS; `GcsStorage` and `PubSubQueue` for Google Cloud
Client	`archive_order`

from abc import ABC, abstractmethod


class Storage(ABC):
    @abstractmethod
    def put(self, key: str, data: str) -> str: ...


class Queue(ABC):
    @abstractmethod
    def publish(self, message: str) -> str: ...


class S3Storage(Storage):
    def put(self, key: str, data: str) -> str:
        return f"S3 put {key} ({len(data)} bytes)"


class SqsQueue(Queue):
    def publish(self, message: str) -> str:
        return f"SQS publish {message}"


class GcsStorage(Storage):
    def put(self, key: str, data: str) -> str:
        return f"Cloud Storage put {key} ({len(data)} bytes)"


class PubSubQueue(Queue):
    def publish(self, message: str) -> str:
        return f"Pub/Sub publish {message}"


class CloudFactory(ABC):
    @abstractmethod
    def create_storage(self) -> Storage: ...

    @abstractmethod
    def create_queue(self) -> Queue: ...


class AwsFactory(CloudFactory):
    def create_storage(self) -> Storage:
        return S3Storage()

    def create_queue(self) -> Queue:
        return SqsQueue()


class GcpFactory(CloudFactory):
    def create_storage(self) -> Storage:
        return GcsStorage()

    def create_queue(self) -> Queue:
        return PubSubQueue()


def archive_order(cloud: CloudFactory, order_id: str, receipt: str) -> None:
    storage = cloud.create_storage()
    queue = cloud.create_queue()
    print(storage.put(f"receipts/{order_id}.txt", receipt))
    print(queue.publish(f"order-archived {order_id}"))


archive_order(AwsFactory(), "A-1042", "2 items, 15.00")
archive_order(GcpFactory(), "A-1042", "2 items, 15.00")

OutputRan with python:3.12-slim (Python 3.12.13)

S3 put receipts/A-1042.txt (14 bytes)
SQS publish order-archived A-1042
Cloud Storage put receipts/A-1042.txt (14 bytes)
Pub/Sub publish order-archived A-1042

`archive_order` asks the factory for both products and names neither provider.

The classes return strings instead of calling real cloud services, so the listing runs anywhere; a real version would wrap each provider's client library in the same classes. archive_order never mentions AWS or Google Cloud. The program decides the provider once, typically from configuration at start-up, by constructing one factory and passing it in.

The family guarantee is the reason to use the pattern rather than two separate factory methods. With one factory per provider, the code that archives an order cannot pair an S3 bucket with a Pub/Sub queue, because no factory returns that pair. A third provider is a third factory and its two products, with no change to archive_order.

← → move through sections and cards