Factory functions compose objects by wiring dependencies together. They contain zero business logic - only constructor calls.
Owner: python-architecture-assistant
Applies when: a Python factory function (in factory.py or named create_* / make_*) contains business logic — loops over domain data, conditional dispatch on runtime state, inline function implementations, validation logic, etc. — instead of pure dependency-wiring calls (constructor invocations and object-tree composition only).
Enforcement: rules/python/zero-business-logic-in-factories.yml flags function_definition nodes named create_* / make_* whose body contains a for_statement, if_statement, lambda, or nested function_definition. Config-level conditionals also fire — the agent distinguishes domain logic from wiring-time config branching.
Why: Factories are the application's composition root — they answer "how do you build the object graph?". Mixing in business logic conflates two concerns: object lifecycle (a wiring concern) and domain decisions (a service concern). Tests of the business logic then need to construct the full factory state; refactors of the wiring propagate to business code. Keeping factories to pure constructor calls (and at most config-level conditionals like "if test_mode, use FakeDB else RealDB") makes them obvious to read, trivially testable as composition graphs, and a clean boundary between "what gets built" and "what it does."
Factories should only:
- Pass dependencies to constructors
- Build nested object trees
- Return typed objects (Protocol implementations)
Factories must NOT:
- Contain loops, conditionals, or business logic
- Have inline function implementations with logic
- Mix object creation with execution
Signals to extract factory function or module:
# [BAD] - Global state scattered in application code
# server.py
_client: AlertmanagerClient | None = None
def handle_request():
global _client
if _client is None:
_client = AlertmanagerClient(get_config())
return _client.get_alerts()
# [GOOD] - Extracted to factory.py
# factory.py
_client: AlertmanagerClient | None = None
def get_client() -> AlertmanagerClient:
"""Get or create the Alertmanager client singleton."""
global _client
if _client is None:
logger.debug("Initializing Alertmanager client")
_client = AlertmanagerClient(get_config())
return _client
# server.py
from .factory import get_client
def handle_request():
return get_client().get_alerts()# [BAD] - Repeated initialization logic
# Multiple modules checking if initialized
if not database:
database = Database(config)
if not api_client:
api_client = ApiClient(config)
# [GOOD] - Centralized in factory
def get_database() -> Database:
global _database
if _database is None:
_database = Database(get_config())
return _database# [BAD] - Complex wiring in main.py (>10 lines of composition)
# main.py
def main():
# ... 20+ lines of dependency wiring ...
# [GOOD] - Extract to factory functions
# factory.py
def create_user_service(database: Database) -> UserService:
return UserService(
repo=SqlUserRepository(database),
logger=ConsoleLogger(),
validator=UserValidator(),
)
# main.py
def main():
service = create_user_service(database)Don't extract for:
- Simple 1-2 line object creation
- Objects with no dependencies
- Code that only creates object once in main.py
# [GOOD] - No factory needed for simple cases
# main.py
def main():
logger = ConsoleLogger() # Simple, no extraction needed
config = Config.from_env()Reference: alertmanager-mcp migrated global _client state from server.py to factory.py, demonstrating when extraction improves code organization.
Services/Applications:
pkg/factory.py # All factory functions in ONE file
Or inline in main.py for simpler applications.
Naming:
- Factories:
create_*prefix (e.g.,create_user_service) - Constructors: Class
__init__(e.g.,UserService(...))
def create_user_service(database: Database) -> UserService:
return UserService(
repo=SqlUserRepository(database),
logger=ConsoleLogger(),
validator=UserValidator(),
)def create_message_handler(database: Database, producer: Producer) -> MessageHandler:
return RetryMessageHandler(
MetricsMessageHandler(
LoggingMessageHandler(
UserMessageHandler(
repo=SqlUserRepository(database),
sender=KafkaSender(producer),
),
logger=ConsoleLogger(),
),
metrics=PrometheusMetrics(),
),
max_retries=3,
)def create_order_fetcher(api_client: ApiClient, queue: QueueClient, branch: str) -> OrderFetcher:
return OrderFetcher(
api_client=api_client,
sender=OrderSender(
queue_client=queue,
topic=f"{branch}-orders",
),
)def create_validators() -> list[Validator]:
return [
EmailValidator(),
PhoneValidator(),
AddressValidator(CountryLookup()),
]def create_command_handler(
database: Database,
producer: Producer,
branch: str,
) -> CommandHandler:
return CommandHandler(
user_fetcher=create_user_fetcher(database, producer, branch),
order_fetcher=create_order_fetcher(database, producer, branch),
account_fetcher=create_account_fetcher(database, producer, branch),
)# BAD: Loop, error handling, conditionals in factory
def create_batch_handler(database: Database, index: Index) -> MessageHandler:
def handle_batch(messages: list[Message]) -> None:
batch = index.new_batch()
for message in messages: # Loop = business logic!
try:
user = parse_user(message)
batch.add(user)
except Exception as e:
logging.error(f"Failed: {e}") # Error handling!
if batch.size() > 0: # Conditional!
index.commit(batch)
return MessageHandler(handle_batch)Why bad:
- For loop iterating messages
- Try/except error handling
- Conditional logic
- Inline function with behavior
# In factory.py:
def create_batch_handler(database: Database, index: Index) -> MessageHandler:
return BatchIndexHandler(index, SqlUserRepository(database))
# In batch_index_handler.py:
class BatchIndexHandler:
def __init__(self, index: Index, repo: UserRepository):
self._index = index
self._repo = repo
def handle(self, messages: list[Message]) -> None:
batch = self._index.new_batch()
for message in messages:
try:
user = self._repo.parse(message)
batch.add(user)
except Exception as e:
logging.error(f"Failed: {e}")
if batch.size() > 0:
self._index.commit(batch)# BAD
def create_service(database: Database) -> UserService:
service = UserService(database)
service.initialize() # Execution!
service.warm_cache() # More execution!
return service# BAD
_instance: UserService | None = None
def create_user_service(database: Database) -> UserService:
global _instance
if _instance is None: # Conditional!
_instance = UserService(database)
return _instance# BAD
def create_repository(config: Config) -> UserRepository:
if config.use_cache: # Conditional!
return CachedUserRepository(SqlUserRepository())
else:
return SqlUserRepository()# GOOD
def create_repository(database: Database) -> UserRepository:
return SqlUserRepository(database)
def create_cached_repository(database: Database, cache: Cache) -> UserRepository:
return CachedUserRepository(
repo=SqlUserRepository(database),
cache=cache,
)Factories wire the application together:
def main():
# 1. Infrastructure
database = Database(os.getenv('DATABASE_URL'))
producer = KafkaProducer(os.getenv('KAFKA_BROKERS'))
branch = os.getenv('BRANCH')
# 2. Use factories to build services
user_service = create_user_service(database)
command_handler = create_command_handler(database, producer, branch)
# 3. Build server
server = HttpServer(
port=int(os.getenv('PORT', '8080')),
handler=create_http_handler(user_service),
)
# 4. Run
server.run()Use factory functions when:
- Same composition used in multiple places
- Complex nested composition (3+ levels)
- Testing needs different wiring
- Composition is getting long in main.py
Use inline wiring when:
- Simple 1-2 level composition
- Used only once
- Easy to read inline
# Inline is fine for simple cases
def main():
user_service = UserService(
repo=SqlUserRepository(database),
logger=ConsoleLogger(),
)# BAD
pkg/user_factory.py
pkg/order_factory.py
pkg/handler_factory.py
# GOOD
pkg/factory.py # All factories in one file# BAD
def create_service(database: Database) -> UserService:
logging.info("Creating service...") # Side effect!
return UserService(database)# BAD - usually unnecessary
async def create_service(database: Database) -> UserService:
await database.ping() # Execution in factory!
return UserService(database)
# GOOD - async in the service, not factory
def create_service(database: Database) -> UserService:
return UserService(database) # Service handles async internallyFactory Checklist:
- All factories in single file:
pkg/factory.pyor inline inmain.py - Use
create_*prefix - Only constructor calls - zero business logic
- No loops, conditionals, or error handling
- Move complex logic to separate implementation classes
- Return typed objects (Protocol implementations preferred)
- python-architecture-patterns.md - Service architecture overview
- python-ioc-guide.md - Dependency injection patterns
- go-factory-pattern.md - Equivalent patterns in Go