Factory Pattern Explained: Simple Factory, Factory Method & Abstract Factory

The Problem

You're building a system that supports multiple databases — MySQL, PostgreSQL, MongoDB. Different services in your application need to create database-specific objects — queries, transactions, updators.

The naive approach? Let each service decide which object to create:

from abc import ABC, abstractmethod

class Database(ABC):
    @abstractmethod
    def create_connection(self): pass

    @abstractmethod
    def create_connection_pool(self): pass

    @abstractmethod
    def replication(self): pass


class Mysql(Database):
    def create_connection(self):
        print("Creating database connection for MySQL")

    def create_connection_pool(self):
        print("Creating database connection pool for MySQL")

    def replication(self):
        print("Managing replication of MySQL database")


class PgSQL(Database):
    def create_connection(self):
        print("Creating database connection for PostgreSQL")

    def create_connection_pool(self):
        print("Creating database connection pool for PostgreSQL")

    def replication(self):
        print("Managing replication of PostgreSQL database")


class MongoDB(Database):
    def create_connection(self):
        print("Creating database connection for MongoDB")

    def create_connection_pool(self):
        print("Creating database connection pool for MongoDB")

    def replication(self):
        print("Managing replication of MongoDB database")


class Query(ABC): pass
class SqlQuery(Query): pass
class MongoQuery(Query): pass


class UserService:
    def create_user(self, db: Database):
        if isinstance(db, Mysql):
            return SqlQuery()
        if isinstance(db, PgSQL):
            return SqlQuery()
        if isinstance(db, MongoDB):
            return MongoQuery()
        return None

Look at UserService.create_user. It has two jobs:

1. Creating a user
2. Deciding which query object to instantiate based on the database type

That's a Single Responsibility Principle violation — create_user should only create users, not make infrastructure decisions.

And every time you add a new database — say Cassandra — you open UserService and add another if. That's an Open/Closed Principle violation — you're modifying existing, working code to add new behaviour.

This is where the Factory pattern comes in.

---

Step 1: Simple Factory

The first fix is straightforward — extract the object creation logic into its own class. UserService delegates to it:

class DatabaseFactory:
    @staticmethod
    def create_query(db: Database):
        if isinstance(db, Mysql):
            return SqlQuery()
        if isinstance(db, PgSQL):
            return SqlQuery()
        if isinstance(db, MongoDB):
            return MongoQuery()
        return None


class UserService:
    def create_user(self, db: Database):
        return DatabaseFactory.create_query(db)

UserService is now clean — one responsibility. It doesn't know or care which query type gets created.

What Simple Factory gives you:

• ✅ SRP fixed — UserService has one job
• ✅ Centralised object creation — one place to look

What Simple Factory doesn't give you:

• ❌ OCP still violated — adding Cassandra means opening DatabaseFactory and adding another if

The if block didn't disappear. It just moved from UserService to DatabaseFactory. We need to go further.

---

Step 2: Factory Method

The real fix is to eliminate the if block entirely. Instead of a separate factory deciding which query to create — each database knows how to create its own query.

Add create_query as an abstract method on Database. Each subclass implements it and returns the right query for its family:

class Database(ABC):
    @abstractmethod
    def create_connection(self): pass

    @abstractmethod
    def create_connection_pool(self): pass

    @abstractmethod
    def replication(self): pass

    @abstractmethod
    def create_query(self): pass  # ← the factory method


class Mysql(Database):
    def create_connection(self):
        print("Creating database connection for MySQL")

    def create_connection_pool(self):
        print("Creating database connection pool for MySQL")

    def replication(self):
        print("Managing replication of MySQL database")

    def create_query(self):
        return SqlQuery()  # MySQL knows it needs SqlQuery


class PgSQL(Database):
    def create_connection(self):
        print("Creating database connection for PostgreSQL")

    def create_connection_pool(self):
        print("Creating database connection pool for PostgreSQL")

    def replication(self):
        print("Managing replication of PostgreSQL database")

    def create_query(self):
        return SqlQuery()  # PostgreSQL also uses SqlQuery


class MongoDB(Database):
    def create_connection(self):
        print("Creating database connection for MongoDB")

    def create_connection_pool(self):
        print("Creating database connection pool for MongoDB")

    def replication(self):
        print("Managing replication of MongoDB database")

    def create_query(self):
        return MongoQuery()  # MongoDB knows it needs MongoQuery


class Query(ABC): pass
class SqlQuery(Query): pass
class MongoQuery(Query): pass


class UserService:
    def create_user(self, db: Database):
        return db.create_query()  # no isinstance, no if blocks

No isinstance. No if blocks. UserService just calls db.create_query() and polymorphism handles the rest.

Adding Cassandra tomorrow? Create a new Cassandra class, implement create_query(), done. Zero changes to UserService or any existing class. OCP satisfied.

class CassandraQuery(Query): pass

class Cassandra(Database):
    def create_connection(self):
        print("Creating database connection for Cassandra")

    def create_connection_pool(self):
        print("Creating database connection pool for Cassandra")

    def replication(self):
        print("Managing replication of Cassandra database")

    def create_query(self):
        return CassandraQuery()

# UserService works with zero changes
print(type(UserService().create_user(Cassandra())).__name__)  # CassandraQuery

What Factory Method gives you:

• ✅ SRP fixed
• ✅ OCP satisfied — new databases don't touch existing code
• ✅ No if blocks — polymorphism does the work

What Factory Method doesn't give you:

• ❌ What if each database needs to create not just a query, but also a transaction handler, an updator, a migration runner? Dumping all of these as abstract methods on Database creates a fat interface — an Interface Segregation Principle violation. Classes that don't support transactions are forced to implement them anyway.

---

Step 3: Abstract Factory

When you have a family of related objects that must be consistent with each other — query, transaction, updator all from the same database ecosystem — Abstract Factory is the answer.

Instead of adding more abstract methods to Database, we give each database one method: create_factory(). It returns a factory specific to that database. The factory then produces the entire family of related objects:

from abc import ABC, abstractmethod


# ── Product families ──────────────────────────────────────────

class Query(ABC): pass
class SqlQuery(Query): pass
class MongoQuery(Query): pass

class Transaction(ABC): pass
class SqlTransaction(Transaction): pass
class NoSqlTransaction(Transaction): pass

class Updator(ABC): pass
class SqlUpdator(Updator): pass
class NoSqlUpdator(Updator): pass


# ── Abstract Factory ──────────────────────────────────────────

class AbstractFactory(ABC):
    @abstractmethod
    def create_query(self): pass

    @abstractmethod
    def create_transaction(self): pass

    @abstractmethod
    def create_updator(self): pass


# ── Concrete Factories ────────────────────────────────────────

class MysqlFactory(AbstractFactory):
    def create_query(self):
        return SqlQuery()

    def create_transaction(self):
        return SqlTransaction()

    def create_updator(self):
        return SqlUpdator()


class PgsqlFactory(AbstractFactory):
    def create_query(self):
        return SqlQuery()

    def create_transaction(self):
        return SqlTransaction()

    def create_updator(self):
        return SqlUpdator()


class MongodbFactory(AbstractFactory):
    def create_query(self):
        return MongoQuery()

    def create_transaction(self):
        return NoSqlTransaction()

    def create_updator(self):
        return NoSqlUpdator()


# ── Database hierarchy ────────────────────────────────────────

class Database(ABC):
    @abstractmethod
    def create_connection(self): pass

    @abstractmethod
    def create_connection_pool(self): pass

    @abstractmethod
    def replication(self): pass

    @abstractmethod
    def create_factory(self): pass  # ← one method, returns the whole family


class Mysql(Database):
    def create_connection(self):
        print("Creating database connection for MySQL")

    def create_connection_pool(self):
        print("Creating database connection pool for MySQL")

    def replication(self):
        print("Managing replication of MySQL database")

    def create_factory(self):
        return MysqlFactory()


class PgSQL(Database):
    def create_connection(self):
        print("Creating database connection for PostgreSQL")

    def create_connection_pool(self):
        print("Creating database connection pool for PostgreSQL")

    def replication(self):
        print("Managing replication of PostgreSQL database")

    def create_factory(self):
        return PgsqlFactory()


class MongoDB(Database):
    def create_connection(self):
        print("Creating database connection for MongoDB")

    def create_connection_pool(self):
        print("Creating database connection pool for MongoDB")

    def replication(self):
        print("Managing replication of MongoDB database")

    def create_factory(self):
        return MongodbFactory()


# ── Services ──────────────────────────────────────────────────

class UserService:
    def create_user(self, db: Database):
        factory = db.create_factory()
        return factory.create_query()


class TransactionService:
    def create_transaction(self, db: Database):
        factory = db.create_factory()
        return factory.create_transaction()


class UpdaterService:
    def create_updator(self, db: Database):
        factory = db.create_factory()
        return factory.create_updator()

Every service gets a factory from the database, then asks the factory for what it needs. The factory guarantees consistency — you'll never accidentally get a MongoQuery paired with a SqlTransaction.

Adding a New Database Family

Cassandra comes along. Zero changes to existing code:

class CassandraQuery(Query): pass
class CassandraTransaction(Transaction): pass
class CassandraUpdator(Updator): pass

class CassandraFactory(AbstractFactory):
    def create_query(self):
        return CassandraQuery()

    def create_transaction(self):
        return CassandraTransaction()

    def create_updator(self):
        return CassandraUpdator()

class Cassandra(Database):
    def create_connection(self):
        print("Creating database connection for Cassandra")

    def create_connection_pool(self):
        print("Creating database connection pool for Cassandra")

    def replication(self):
        print("Managing replication of Cassandra database")

    def create_factory(self):
        return CassandraFactory()


# All services work immediately — zero existing code touched
print(type(UserService().create_user(Cassandra())).__name__)          # CassandraQuery
print(type(TransactionService().create_transaction(Cassandra())).__name__)  # CassandraTransaction
print(type(UpdaterService().create_updator(Cassandra())).__name__)    # CassandraUpdator

---

The Progression — Why Each Pattern Exists

Pattern	What it solved	What it left broken
No pattern — isinstance in UserService	Nothing	SRP + OCP violated
Simple Factory — DatabaseFactory class	SRP fixed in UserService	OCP still violated inside factory
Factory Method — create_query on each DB	OCP fixed, no if blocks	Fat interface — ISP violated as methods grow
Abstract Factory — family of factories	ISP fixed, families consistent	Nothing — clean ✅

Each pattern solves the problem the previous one left behind. This is the progression you apply in the real world — start simple, upgrade when the pain demands it.

---

When to Use Each

Simple Factory — when object creation logic is straightforward and new types are unlikely. Centralises creation but accepts the OCP tradeoff. Good for small, stable systems.

Factory Method — when you have one product to create per type and want to eliminate if/else through polymorphism. Each subclass decides what it creates. Best when each class in the hierarchy creates exactly one related object.

Abstract Factory — when you have a family of related objects that must be consistent with each other, and new families might be introduced. Guarantees that objects from the same family always work together correctly.

One product per type?
    └── Factory Method

A family of related products per type?
    └── Abstract Factory

Simple, unlikely to change?
    └── Simple Factory

---

The Tradeoffs

Simple Factory

• ✅ Easy to understand, centralised creation
• ❌ OCP violated — every new type modifies the factory
• ❌ Factory grows indefinitely as types are added

Factory Method

• ✅ OCP satisfied — new types don't touch existing code
• ✅ No if blocks — polymorphism handles dispatch
• ❌ Fat interface risk — adding more products pollutes the base class
• ❌ Each subclass tightly coupled to its product type

Abstract Factory

• ✅ Families are consistent — no accidental mixing of SQL and NoSQL objects
• ✅ OCP fully satisfied — new families are entirely new classes
• ✅ ISP respected — Database has one factory method, not many product methods
• ❌ Most complex — more classes, more files
• ❌ Adding a new product to the family (e.g. create_migration) requires changes to every concrete factory — OCP violation within the family
• ❌ Overkill for simple systems with one or two object types

---

The SOLID Connection

Factory patterns are SOLID in action:

SRP — UserService creates users. DatabaseFactory or the concrete factory creates objects. One responsibility each.

OCP — Factory Method and Abstract Factory let you add new database types without touching existing classes. New family = new classes only.

LSP — UserService works with any Database subclass interchangeably. TransactionService works with any factory that implements AbstractFactory. Substitution guaranteed.

ISP — Abstract Factory avoids dumping all product creation methods on Database. One method — create_factory() — returns a focused factory interface.

DIP — UserService depends on Database and AbstractFactory abstractions, not on Mysql or MysqlFactory concretions. Swap the database, the service never changes.