# Architecture

> Applies SOLID principles, clean architecture layers, dependency injection, and patterns like CQRS and event sourcing to design scalable, maintainable systems.

- Skill: `neuralblitz/architecture` (Agent Skill)
- Install (CLI): `npx skillmds@latest add neuralblitz/architecture`
- Raw SKILL.md: https://api.skillmd.com/api/skills/neuralblitz/architecture/raw
- Safety review: PASS (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools, AI & ML, API Design
- Tags: Clean Architecture, Cqrs, Dependency Injection, Event Sourcing, Python, Solid Principles
- License: MIT
- Author: NeuralBlitz (https://skillmd.com/u/neuralblitz)
- Updated: 2026-08-22
- Page: https://skillmd.com/skills/neuralblitz/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
```python
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
```python
# 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
```python
# 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]
```