Data/ML Engineer Blog

Factory Method (pick which object to create in a subclass)

When to use

  • You have a base workflow, but creation of one part varies (e.g., Snowflake auth: key pair vs OAuth).
  • You want to add new creation modes without touching the base workflow.
  • You need testable swaps for the created thing.

Avoid when a plain function like make_connection(config) is enough.

Diagram (text)

SnowflakeService (base)
 ├─ execute(sql) uses → create_conn()  ← Factory Method (overridden)
 ├─ (abstract) create_conn(): SnowflakeConn
 └─▲
   ├── KeyPairService → returns KeyPairConn
   └── OAuthService   → returns OAuthConn

Step-by-step idea

  1. Put the shared workflow in the base class (execute).
  2. Make an abstract factory method in the base (create_conn).
  3. Each subclass overrides that method to build the right object.
  4. The base code calls create_conn() and stays agnostic.

Python example (≤40 lines, type-hinted)

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol

class SnowflakeConn(Protocol):
    def run(self, sql: str) -> str: ...

@dataclass
class KeyPairConn:
    user: str; key_path: str
    def run(self, sql: str) -> str: return f"keypair:{self.user}:{sql}"

@dataclass
class OAuthConn:
    token: str
    def run(self, sql: str) -> str: return f"oauth:{len(self.token)}:{sql}"

class SnowflakeService(ABC):
    def execute(self, sql: str) -> str:
        conn = self.create_conn()
        return conn.run(sql)
    @abstractmethod
    def create_conn(self) -> SnowflakeConn: ...

@dataclass
class KeyPairService(SnowflakeService):
    user: str; key_path: str
    def create_conn(self) -> SnowflakeConn: return KeyPairConn(self.user, self.key_path)

@dataclass
class OAuthService(SnowflakeService):
    token: str
    def create_conn(self) -> SnowflakeConn: return OAuthConn(self.token)

Tiny pytest (cements it)

def test_factory_method_switches_impl():
    svc1 = KeyPairService(user="svc", key_path="/keys/id_rsa")
    svc2 = OAuthService(token="abc123")
    assert "keypair:svc" in svc1.execute("SELECT 1")
    assert "oauth:" in svc2.execute("SELECT 1")

Trade-offs & pitfalls

  • Pros: Shared workflow in one place; easy to add new connection types; good test seams.
  • Cons: Extra classes; indirection may feel heavy for simple cases.
  • Pitfalls:
    • Putting business logic inside create_conn—keep it just for creation.
    • Returning objects with incompatible interfaces—stick to a Protocol (here SnowflakeConn).
    • Overusing inheritance—if you have many tiny subclasses, consider a map of callables.

Pythonic alternatives

  • Simple factory function: def make_conn(cfg) -> SnowflakeConn: (often enough).
  • Constructor injection: Pass a callable conn_factory: Callable[[], SnowflakeConn] into one service class.
  • Dataclass + registry: REGISTRY = {"keypair": lambda cfg: KeyPairConn(...), "oauth": ...}.
  • Pydantic/attrs for config validation before calling the factory.

Mini exercise

Add PasswordService(user, password) that returns a PasswordConn. Confirm your existing execute() needs no changes and the new service passes a small test.

Checks (quick checklist)

  • Base class holds the workflow; subclasses only decide what to create.
  • Factory method returns a stable interface/Protocol.
  • Adding a new product doesn’t touch base code.
  • Tests verify different subclasses produce different behaviors with the same execute.
  • Creation logic is thin; business logic stays in the workflow.