CreationalPythonverifiedVerified

Abstract Factory Pattern in Python

Provides an interface for creating families of related objects without specifying their concrete classes.

How to Implement the Abstract Factory Pattern in Python

1Step 1: Define the product interfaces

from abc import ABC, abstractmethod


class Button(ABC):
    @abstractmethod
    def render(self) -> str: ...


class TextInput(ABC):
    @abstractmethod
    def render(self) -> str: ...

2Step 2: Define the abstract factory interface

class UIFactory(ABC):
    @abstractmethod
    def create_button(self) -> Button: ...

    @abstractmethod
    def create_text_input(self) -> TextInput: ...

3Step 3: Implement concrete product families

class DarkButton(Button):
    def render(self) -> str:
        return "<button class='dark-btn'>Click</button>"


class DarkTextInput(TextInput):
    def render(self) -> str:
        return "<input class='dark-input' />"


class DarkThemeFactory(UIFactory):
    def create_button(self) -> Button:
        return DarkButton()

    def create_text_input(self) -> TextInput:
        return DarkTextInput()


class LightButton(Button):
    def render(self) -> str:
        return "<button class='light-btn'>Click</button>"


class LightTextInput(TextInput):
    def render(self) -> str:
        return "<input class='light-input' />"


class LightThemeFactory(UIFactory):
    def create_button(self) -> Button:
        return LightButton()

    def create_text_input(self) -> TextInput:
        return LightTextInput()

4Step 4: Build forms using any factory

def build_form(factory: UIFactory) -> str:
    button = factory.create_button()
    text_input = factory.create_text_input()
    return f"Form: {text_input.render()} {button.render()}"


print(build_form(DarkThemeFactory()))
print(build_form(LightThemeFactory()))

"""Cross-database Abstract Factory with connection and query builder products."""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any


# [step] Define product interfaces for connection and query builder
@dataclass
class QueryResult:
    rows: list[dict[str, Any]]
    row_count: int


class DbConnection(ABC):
    @property
    @abstractmethod
    def dialect(self) -> str: ...

    @abstractmethod
    async def connect(self, host: str, port: int, database: str) -> None: ...

    @abstractmethod
    async def disconnect(self) -> None: ...

    @abstractmethod
    def is_connected(self) -> bool: ...


class QueryBuilder(ABC):
    @abstractmethod
    def select(self, table: str, columns: list[str] | None = None) -> "QueryBuilder": ...

    @abstractmethod
    def where(self, condition: str, params: list[Any] | None = None) -> "QueryBuilder": ...

    @abstractmethod
    def limit(self, count: int) -> "QueryBuilder": ...

    @abstractmethod
    def build(self) -> tuple[str, list[Any]]: ...

    @abstractmethod
    async def execute(self) -> QueryResult: ...


class DatabaseFactory(ABC):
    @abstractmethod
    def create_connection(self) -> DbConnection: ...

    @abstractmethod
    def create_query_builder(self, conn: DbConnection) -> QueryBuilder: ...


# [step] Implement PostgreSQL product family
class PgConnection(DbConnection):
    def __init__(self) -> None:
        self._connected = False

    @property
    def dialect(self) -> str:
        return "postgresql"

    async def connect(self, host: str, port: int, database: str) -> None:
        self._connected = True

    async def disconnect(self) -> None:
        self._connected = False

    def is_connected(self) -> bool:
        return self._connected


class PgQueryBuilder(QueryBuilder):
    def __init__(self, conn: DbConnection) -> None:
        self._conn = conn
        self._table = ""
        self._columns: list[str] = ["*"]
        self._conditions: list[str] = []
        self._params: list[Any] = []
        self._limit_count: int | None = None

    def select(self, table: str, columns: list[str] | None = None) -> "PgQueryBuilder":
        self._table = table
        self._columns = columns or ["*"]
        return self

    def where(self, condition: str, params: list[Any] | None = None) -> "PgQueryBuilder":
        self._conditions.append(condition)
        self._params.extend(params or [])
        return self

    def limit(self, count: int) -> "PgQueryBuilder":
        self._limit_count = count
        return self

    def build(self) -> tuple[str, list[Any]]:
        sql = f"SELECT {', '.join(self._columns)} FROM {self._table}"
        if self._conditions:
            sql += f" WHERE {' AND '.join(self._conditions)}"
        if self._limit_count is not None:
            sql += f" LIMIT {self._limit_count}"
        return sql, self._params

    async def execute(self) -> QueryResult:
        if not self._conn.is_connected():
            raise RuntimeError("Not connected")
        return QueryResult(rows=[], row_count=0)


class PostgresFactory(DatabaseFactory):
    def create_connection(self) -> DbConnection:
        return PgConnection()

    def create_query_builder(self, conn: DbConnection) -> QueryBuilder:
        return PgQueryBuilder(conn)


# [step] Implement SQLite product family
class SqliteConnection(DbConnection):
    def __init__(self) -> None:
        self._connected = False

    @property
    def dialect(self) -> str:
        return "sqlite"

    async def connect(self, host: str, port: int, database: str) -> None:
        self._connected = True

    async def disconnect(self) -> None:
        self._connected = False

    def is_connected(self) -> bool:
        return self._connected


class SqliteQueryBuilder(QueryBuilder):
    def __init__(self, conn: DbConnection) -> None:
        self._conn = conn
        self._table = ""
        self._columns: list[str] = ["*"]
        self._conditions: list[str] = []
        self._params: list[Any] = []
        self._limit_count: int | None = None

    def select(self, table: str, columns: list[str] | None = None) -> "SqliteQueryBuilder":
        self._table = table
        self._columns = columns or ["*"]
        return self

    def where(self, condition: str, params: list[Any] | None = None) -> "SqliteQueryBuilder":
        self._conditions.append(condition)
        self._params.extend(params or [])
        return self

    def limit(self, count: int) -> "SqliteQueryBuilder":
        self._limit_count = count
        return self

    def build(self) -> tuple[str, list[Any]]:
        sql = f"SELECT {', '.join(self._columns)} FROM {self._table}"
        if self._conditions:
            sql += f" WHERE {' AND '.join(self._conditions)}"
        if self._limit_count is not None:
            sql += f" LIMIT {self._limit_count}"
        return sql, self._params

    async def execute(self) -> QueryResult:
        if not self._conn.is_connected():
            raise RuntimeError("Not connected")
        return QueryResult(rows=[], row_count=0)


class SqliteFactory(DatabaseFactory):
    def create_connection(self) -> DbConnection:
        return SqliteConnection()

    def create_query_builder(self, conn: DbConnection) -> QueryBuilder:
        return SqliteQueryBuilder(conn)


# [step] Factory registry for runtime selection
_FACTORIES: dict[str, type[DatabaseFactory]] = {
    "postgresql": PostgresFactory,
    "sqlite": SqliteFactory,
}


def get_database_factory(dialect: str) -> DatabaseFactory:
    cls = _FACTORIES.get(dialect)
    if cls is None:
        raise ValueError(f'Unsupported dialect "{dialect}"')
    return cls()

Abstract Factory Pattern Architecture

hourglass_empty

Rendering diagram...

lightbulb

Abstract Factory Pattern in the Real World

“Imagine furnishing a room from IKEA versus a luxury boutique. Each store (factory) produces a complete set of furniture — chairs, tables, sofas — that share a consistent style. You pick the store, and every piece you get matches. You never mix a rustic IKEA chair with a baroque boutique table.”