Determines the correct architectural layer (Entry Point/Service/Domain/Infrastructure) for code placement in Python applications using Service Layer pattern...
Where should this code go?
โ
โผ
Does it interact with users/callers?
(Parse args, format output, routes)
โ
โโ YES โ ENTRY POINT (CLI, API routes, scripts)
โ
โโ NO โ Does it call external systems? (API, DB, FS)
โ
โโ YES โ INFRASTRUCTURE (API clients, DB, file I/O, git)
โ
โโ NO โ Is it pure data or business rules?
โ
โโ YES โ DOMAIN (models, dataclasses, parsing, validation)
โ
โโ NO โ SERVICE (business logic, coordination)
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Entry Point Layer โ โ User/caller interface
โ (CLI, API routes, scripts, handlers) โ (orchestration only)
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Service Layer โ โ Business logic
โ Core: Single-responsibility services โ (coordinates operations)
โ Composite: Multi-service coordination โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Infrastructure Layer โ โ External integrations
โ (API clients, DB, filesystem, git) โ (I/O boundaries)
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Domain Layer โ โ Pure business models
โ (dataclasses, models, parsing, rules) โ (no I/O, no dependencies)
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
Dependency flow:
Entry Point โ Service โ Infrastructure
Entry Point โ Service โ Domain
Domain: No dependencies (pure)
Purpose: Interface between users/callers and the application
Belongs here:
File locations: cli/, api/, scripts/, main.py, __main__.py
Note: For detailed command dispatcher patterns and CLI architecture, see the cli-architecture skill.
Example:
def cmd_prepare(args, gh):
"""CLI command orchestrates services."""
# 1. Get configuration
repo = os.environ.get("GITHUB_REPOSITORY")
# 2. Initialize infrastructure
metadata_store = GitHubMetadataStore(repo)
# 3. Initialize services
metadata_service = MetadataService(metadata_store)
task_service = TaskService(repo, metadata_service)
# 4. Use services
task = task_service.find_next_available_task(spec_content)
# 5. Format output for user
print(f"Next task: {task.title}")
Purpose: Encapsulate business logic and coordinate operations
Belongs here:
File locations: services/, business/
Note: For service constructor patterns and dependency management, see the dependency-injection skill.
Example - Core service:
class TaskService:
def __init__(self, repo: str, metadata_service: MetadataService):
self.repo = repo
self.metadata_service = metadata_service
def find_next_available_task(self, spec: SpecFile) -> Task:
"""Business logic for finding next task."""
completed = self.metadata_service.get_completed_tasks()
return spec.find_next_pending(completed)
Example - Composite service:
class StatisticsService:
def __init__(self, task_service: TaskService, pr_service: PRService):
self.task_service = task_service
self.pr_service = pr_service
def generate_project_stats(self) -> ProjectStats:
"""Coordinate across services."""
tasks = self.task_service.get_all_tasks()
prs = self.pr_service.get_all_prs()
return ProjectStats(tasks, prs)
Purpose: Pure business models and rules with no external dependencies
Belongs here:
File locations: models/, domain/, entities/, config/
Note: For comprehensive domain modeling guidance and the parse-once principle, see the domain-modeling skill.
Example:
@dataclass
class Task:
"""Pure domain model."""
title: str
status: TaskStatus
assignee: Optional[str] = None
@classmethod
def from_markdown(cls, content: str) -> 'Task':
"""Parse from string (domain parsing)."""
# ... parsing logic ...
return cls(title=title, status=status)
def is_available(self) -> bool:
"""Pure business rule."""
return self.status == TaskStatus.PENDING and self.assignee is None
def validate(self) -> None:
"""Domain validation."""
if not self.title:
raise DomainValidationError("Task must have a title")
Key principle: Domain models parse data once and provide type-safe APIs. No I/O.
Purpose: Wrap external systems and I/O operations
Belongs here:
File locations: infrastructure/, adapters/, clients/
Example:
class GitHubMetadataStore:
"""Infrastructure for GitHub API operations."""
def __init__(self, repo: str):
self.repo = repo
def get_completed_tasks(self) -> List[str]:
"""External I/O: Read from GitHub API."""
result = subprocess.run(
["gh", "api", f"/repos/{self.repo}/issues"],
capture_output=True, text=True
)
data = json.loads(result.stdout)
return [issue['title'] for issue in data if issue['state'] == 'closed']
โ DO NOT put business logic in entry points
# BAD: Business logic in CLI
def cmd_prepare(args):
for task in spec.tasks: # Business logic here
if task.status == "pending":
return task
โ DO delegate to services
# GOOD: Orchestrate only
def cmd_prepare(args):
task = task_service.find_next_available_task(spec_content)
print(f"Next task: {task.title}")
โ DO NOT directly call infrastructure (e.g., subprocess.run(["git", "status"]))
โ
DO use services (git_service.get_status())
โ DO NOT parse arguments or access environment
# BAD: Service reads environment
class TaskService:
def __init__(self):
self.repo = os.environ.get("GITHUB_REPOSITORY")
โ DO receive configuration via constructor
# GOOD: Injected configuration
class TaskService:
def __init__(self, repo: str, metadata_service: MetadataService):
self.repo = repo
self.metadata_service = metadata_service
โ DO NOT make direct subprocess/API calls (use injected infrastructure instead)
โ DO NOT perform I/O operations
# BAD: Domain reads files
@dataclass
class Task:
@classmethod
def from_file(cls, path: str):
with open(path) as f: # I/O belongs in infrastructure
return cls.from_yaml(f.read())
โ DO parse from strings
# GOOD: Parse string, not file
@dataclass
class Task:
@classmethod
def from_yaml(cls, content: str): # String input
data = yaml.safe_load(content)
return cls(**data)
โ DO NOT depend on services or infrastructure (keep domain pure with no dependencies)
โ DO NOT contain business logic
# BAD: Infrastructure makes business decisions
class GitHubMetadataStore:
def get_next_task(self):
tasks = self._fetch_all_tasks()
for task in tasks: # Business logic
if task['status'] == 'pending':
return task
โ DO provide simple CRUD operations
# GOOD: Just fetch data
class GitHubMetadataStore:
def get_all_tasks(self) -> List[dict]:
result = subprocess.run(["gh", "api", "..."])
return json.loads(result.stdout)
Q: "Generate a summary report of tasks and PRs. Where does this go?"
A: Service Layer - Coordinates multiple data sources with business logic
# Service layer
class ReportService:
def __init__(self, task_service: TaskService, pr_service: PRService):
self.task_service = task_service
self.pr_service = pr_service
def generate_summary(self) -> Report:
tasks = self.task_service.get_all_tasks()
prs = self.pr_service.get_all_prs()
return Report(
total_tasks=len(tasks),
completed_tasks=len([t for t in tasks if t.is_complete()])
)
# Domain layer
@dataclass
class Report:
total_tasks: int
completed_tasks: int
def completion_percentage(self) -> float:
return (self.completed_tasks / self.total_tasks) * 100 if self.total_tasks else 0.0
# Entry point
def cmd_report(args):
report = report_service.generate_summary()
print(f"Completion: {report.completion_percentage():.1f}%")
Q: "User dataclass that converts to JSON. Where does it go?"
A: Domain Layer - Pure data model with serialization
# Domain layer
@dataclass
class User:
id: str
username: str
email: str
password_hash: str
def to_json_dict(self) -> dict:
"""Domain logic: what fields to expose."""
return {
"id": self.id,
"username": self.username,
"email": self.email
# password_hash excluded
}
# Entry point (API route)
@app.route('/users/<user_id>')
def get_user(user_id):
user = user_service.get_user_by_id(user_id)
return jsonify(user.to_json_dict())
Q: "Connect to PostgreSQL database. Where does this go?"
A: Infrastructure Layer - External system integration
# Infrastructure: Repository pattern
class UserRepository:
def __init__(self, db: PostgreSQLDatabase):
self.db = db
def find_by_id(self, user_id: str) -> Optional[User]:
rows = self.db.execute_query("SELECT * FROM users WHERE id = %s", (user_id,))
return User.from_db_row(rows[0]) if rows else None
# Service uses repository
class UserService:
def __init__(self, user_repository: UserRepository):
self.user_repository = user_repository
Q: "Validate email format. Where does this go?"
A: Domain Layer - Pure business rule
@dataclass
class Email:
address: str
def __post_init__(self):
if not self._is_valid_format(self.address):
raise InvalidEmailError(f"Invalid: {self.address}")
@staticmethod
def _is_valid_format(email: str) -> bool:
return bool(re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email))
Q: "Commit, push, create branches. Where does this go?"
A: Infrastructure Layer - Wraps external git commands
# Infrastructure wraps git commands
class GitCommandRunner:
def commit(self, message: str) -> None:
self._run(["commit", "-m", message])
# Service adds business logic
class GitService:
def __init__(self, git_runner: GitCommandRunner):
self.git_runner = git_runner
def start_feature(self, feature_name: str) -> str:
"""Business logic: naming convention."""
branch_name = f"feature/{feature_name}"
self.git_runner.create_branch(branch_name)
return branch_name
| Code Type | Layer | Example |
|---|---|---|
| CLI argument parsing | Entry Point | argparse, route definitions |
| Business workflow | Service | Coordinate operations |
| Data model | Domain | @dataclass, models |
| API client | Infrastructure | HTTP requests, gh CLI |
| JSON formatting | Domain | to_json_dict() |
| Database query | Infrastructure | SQL execution |
| Validation rule | Domain | validate(), is_valid() |
| File I/O | Infrastructure | open(), write() |
| Response formatting | Entry Point | jsonify(), print() |
| Multi-service coordination | Service (Composite) | Statistics, reports |
Default to Service layer if uncertain, refactor later if needed.