| name | architecture |
| description | Software architecture patterns and best practices |
| license | MIT |
| compatibility | opencode |
| metadata | {"audience":"developers","category":"architecture"} |
What I do
- Apply SOLID principles
- Implement clean architecture layers
- Use dependency injection
- Design for scalability and maintainability
- Choose appropriate patterns (CQRS, Event Sourcing, etc.)
- Handle cross-cutting concerns
- Design APIs and contracts
- Manage complexity through modularization
When to use me
When designing system architecture or making architectural decisions.
Clean Architecture Layers
┌─────────────────────────────────────────┐
│ Application Layer │
│ (Use Cases, Services, Interactors) │
├─────────────────────────────────────────┤
│ Domain Layer │
│ (Entities, Value Objects, │
│ Domain Services, Aggregates) │
├─────────────────────────────────────────┤
│ Infrastructure Layer │
│ (Database, External Services, │
│ File System, HTTP Clients) │
├─────────────────────────────────────────┤
│ Presentation Layer │
│ (API Controllers, CLI, Web UI) │
└─────────────────────────────────────────┘
Dependency Rule: Outer layers depend on inner layers,
but inner layers know nothing about outer layers.
Dependency Injection
from abc import ABC, abstractmethod
from typing import Protocol
class UserRepository(Protocol):
def find_by_id(self, id: str) -> User | None: ...
def find_by_email(self, email: str) -> User | None: ...
def save(self, user: User) -> None: ...
class EmailService(Protocol):
def send(self, to: str, subject: str, body: str) -> None: ...
class CreateUserUseCase:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailService,
) -> None:
self.user_repository = user_repository
self.email_service = email_service
def execute(self, input: CreateUserInput) -> User:
existing = self.user_repository.find_by_email(.email)
existing:
UserAlreadyExistsError(.email)
user = User(
email=.email,
name=.name,
)
.user_repository.save(user)
.email_service.send(
to=user.email,
subject=,
body=,
)
user
:
() -> :
.db = db
() -> User | :
row = .db.execute(
,
(,)
).fetchone()
User.from_row(row) row
:
() -> :
() -> Container:
Container(
user_repository=PostgresUserRepository(db),
email_service=SMTPEmailService(),
)
CQRS Pattern
class CreateUserCommand:
def __init__(self, email: str, name: str) -> None:
self.email = email
self.name = name
class CreateUserCommandHandler:
def __init__(self, user_repository: UserRepository) -> None:
self.user_repository = user_repository
def handle(self, command: CreateUserCommand) -> User:
user = User(email=command.email, name=command.name)
self.user_repository.save(user)
return user
class GetUserQuery:
def __init__(self, user_id: str) -> None:
self.user_id = user_id
class GetUserQueryHandler:
def __init__(self, user_read_model: UserReadRepository) -> None:
self.user_read_model = user_read_model
def handle(self, query: GetUserQuery) -> UserDTO:
return self.user_read_model.find_by_id(query.user_id)
:
() -> :
.handlers: [, ] = {}
() -> :
.handlers[command_type] = handler
() -> :
handler = .handlers[(command)]
handler(command)
Event Sourcing
class UserEvent:
def __init__(self, user_id: str, timestamp: datetime) -> None:
self.user_id = user_id
self.timestamp = timestamp
class UserCreatedEvent(UserEvent):
def __init__(self, user_id: str, email: str, name: str) -> None:
super().__init__(user_id, datetime.now())
self.email = email
self.name = name
class UserEmailChangedEvent(UserEvent):
def __init__(self, user_id: str, old_email: str, new_email: str) -> None:
super().__init__(user_id, datetime.now())
self.old_email = old_email
self.new_email = new_email
class UserAggregate:
def __init__(self, user_id: str) -> None:
self.user_id = user_id
self._events: list[UserEvent] = []
._version =
() -> :
._events.append(
UserCreatedEvent(.user_id, email, name)
)
() -> :
old_email = (, , )
._events.append(
UserEmailChangedEvent(.user_id, old_email, new_email)
)
() -> :
._version +=
(event, UserCreatedEvent):
._email = event.email
._name = event.name
(event, UserEmailChangedEvent):
._email = event.new_email
() -> [UserEvent]:
._events.copy()
:
() -> :
db.events.insert(event.to_dict())
() -> [UserEvent]:
rows = db.events.find(aggregate_id=aggregate_id)
[UserEvent.from_row(row) row rows]