NeuralBlitz 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
# Define contracts in inner layers
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: ...
# Domain layer (inner) - depends on abstractions
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(input.email)
if existing:
raise UserAlreadyExistsError(input.email)
user = User(
email=input.email,
name=input.name,
)
self.user_repository.save(user)
self.email_service.send(
to=user.email,
subject="Welcome!",
body=f"Hi {user.name}, welcome to our platform!",
)
return user
# Infrastructure layer (outer) - implements abstractions
class PostgresUserRepository:
def __init__(self, db: Connection) -> None:
self.db = db
def find_by_id(self, id: str) -> User | None:
row = self.db.execute(
"SELECT * FROM users WHERE id = ?",
(id,)
).fetchone()
return User.from_row(row) if row else None
class SMTPEmailService:
def send(self, to: str, subject: str, body: str) -> None:
# Send email via SMTP
pass
# Composition root
def create_container() -> Container:
return Container(
user_repository=PostgresUserRepository(db),
email_service=SMTPEmailService(),
)
CQRS Pattern
# Commands (Write)
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
# Queries (Read)
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)
# Dispatcher
class CommandDispatcher:
def __init__(self, container: Container) -> None:
self.handlers: dict[type, Callable] = {}
def register(self, command_type: type, handler: Callable) -> None:
self.handlers[command_type] = handler
def dispatch(self, command: Command) -> Any:
handler = self.handlers[type(command)]
return handler(command)
Event Sourcing
# Events
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
# Aggregate
class UserAggregate:
def __init__(self, user_id: str) -> None:
self.user_id = user_id
self._events: list[UserEvent] = []
self._version = 0
def create(self, email: str, name: str) -> None:
self._events.append(
UserCreatedEvent(self.user_id, email, name)
)
def change_email(self, new_email: str) -> None:
old_email = getattr(self, '_email', None)
self._events.append(
UserEmailChangedEvent(self.user_id, old_email, new_email)
)
def apply(self, event: UserEvent) -> None:
# Apply event to state
self._version += 1
if isinstance(event, UserCreatedEvent):
self._email = event.email
self._name = event.name
elif isinstance(event, UserEmailChangedEvent):
self._email = event.new_email
def get_events(self) -> list[UserEvent]:
return self._events.copy()
# Event Store
class EventStore:
def append(self, event: UserEvent) -> None:
db.events.insert(event.to_dict())
def get_events(self, aggregate_id: str) -> list[UserEvent]:
rows = db.events.find(aggregate_id=aggregate_id)
return [UserEvent.from_row(row) for row in rows]