# Onion Architecture

## Why I Moved Beyond Simple Layering

After a year of working with layered architecture in the POS system, I noticed the same recurring problem: the domain layer still had implicit knowledge of the database. SQLAlchemy models subclassed from `Base`. Repository classes imported from `sqlalchemy.orm`. Changing the ORM version or switching to a different database meant touching domain code.

Onion architecture pushed the domain completely to the centre — with a strict rule that nothing in the domain layer imports from anything outside it. The infrastructure (databases, HTTP clients, external services) lives at the outermost ring and points inward. The domain never imports from infrastructure.

This was a meaningful change. It made the domain genuinely testable without a database. It made the system's business rules readable without understanding any infrastructure choices.

## Table of Contents

* [What Is Onion Architecture?](#what-is-onion-architecture)
* [The Rings](#the-rings)
* [The Dependency Inversion Principle in Practice](#the-dependency-inversion-principle-in-practice)
* [Practical Example: Auth Service with Onion Structure](#practical-example-auth-service-with-onion-structure)
* [Project Structure](#project-structure)
* [Testing Without a Database](#testing-without-a-database)
* [When Onion Architecture Pays Off](#when-onion-architecture-pays-off)
* [Lessons Learned](#lessons-learned)

***

## What Is Onion Architecture?

Onion architecture (introduced by Jeffrey Palermo) organises code as concentric rings. The inner rings define abstractions; the outer rings implement them. Dependencies always point inward — outer rings can depend on inner rings, but inner rings can never depend on outer rings.

{% @mermaid/diagram content="graph TB
subgraph Onion\["Onion Architecture (cross-section)"]
DOMAIN\["🏛️ Domain<br/>(Entities, Value Objects,<br/>Domain Services)"]
APP\["⚙️ Application<br/>(Use Cases, Application Services,<br/>Repository Interfaces)"]
INFRA\["🔌 Infrastructure<br/>(DB, HTTP, Email, Files)<br/>Implements App interfaces"]
UI\["🖥️ User Interface / API<br/>(Controllers, CLI, Event Consumers)"]
end

```
UI --> INFRA
UI --> APP
INFRA --> APP
APP --> DOMAIN

style DOMAIN fill:#ffeaa7
style APP fill:#81ecec
style INFRA fill:#a29bfe
style UI fill:#fd79a8" %}
```

The key difference from layered architecture: the **database is not at the bottom of the stack**. It is at the outermost ring, alongside HTTP controllers and CLI commands. The domain is in the centre and knows nothing about any of them.

***

## The Rings

### Ring 1: Domain (Centre)

* **Entities** — objects with identity that persist across time (e.g., `User`, `Order`, `Product`)
* **Value Objects** — immutable objects defined by their attributes, not identity (e.g., `Email`, `Money`, `TenantId`)
* **Domain Services** — operations that do not naturally belong to a single entity
* **Domain Events** — facts that have happened within the domain

**Rule:** Zero imports from outside this ring. No ORM base classes, no HTTP types, no third-party libraries.

### Ring 2: Application

* **Use Cases / Application Services** — orchestrate domain objects to fulfil business goals
* **Repository Interfaces** — abstract definitions of persistence operations (not implementations)
* **Service Interfaces** — abstract definitions of external capabilities (email, notifications, etc.)
* **DTOs** — data objects for passing data across ring boundaries

**Rule:** Can import from Domain. Does not import from Infrastructure.

### Ring 3: Infrastructure

* **Repository Implementations** — concrete database implementations of the interfaces defined in Application
* **External Service Clients** — HTTP clients, email providers, file storage
* **ORM Models** — SQLAlchemy classes or equivalent

**Rule:** Can import from Application. Implements Application-defined interfaces. Never imported by Application or Domain.

### Ring 4: User Interface (Outermost)

* **HTTP Controllers / Routers** — FastAPI routers, Flask views
* **CLI Commands**
* **Background Job Handlers**
* **Event Consumers**

**Rule:** Depends on Application through use case classes. Never bypasses Application to call Domain directly.

***

## The Dependency Inversion Principle in Practice

The mechanism that makes the onion work is **dependency inversion**: high-level modules (domain, application) do not depend on low-level modules (infrastructure). Both depend on abstractions (interfaces).

```python
# Application ring: define the interface
# auth/application/repositories.py

from abc import ABC, abstractmethod
from ..domain.models import User

class UserRepository(ABC):
    @abstractmethod
    def find_by_email(self, tenant_id: str, email: str) -> User | None: ...

    @abstractmethod
    def find_by_id(self, tenant_id: str, user_id: int) -> User | None: ...

    @abstractmethod
    def save(self, user: User) -> User: ...
```

```python
# Infrastructure ring: implement the interface
# auth/infrastructure/db/user_repository.py

from sqlalchemy.orm import Session
from ...application.repositories import UserRepository  # Points INWARD to Application
from ...domain.models import User
from .orm_models import UserORM

class SQLUserRepository(UserRepository):
    def __init__(self, db: Session):
        self._db = db

    def find_by_email(self, tenant_id: str, email: str) -> User | None:
        row = (
            self._db.query(UserORM)
            .filter_by(tenant_id=tenant_id, email=email)
            .first()
        )
        return self._to_domain(row) if row else None

    def save(self, user: User) -> User:
        orm = UserORM(
            tenant_id=user.tenant_id,
            email=user.email.value,
            password_hash=user.password_hash,
            role=user.role.value,
        )
        self._db.merge(orm)
        self._db.commit()
        return user

    def _to_domain(self, row: UserORM) -> User:
        from ...domain.models import Email, Role
        return User(
            id=row.id,
            tenant_id=row.tenant_id,
            email=Email(row.email),
            password_hash=row.password_hash,
            role=Role(row.role),
        )
```

The Application ring defines `UserRepository` as an abstract interface. The Infrastructure ring implements it. The Application (use cases) receives the repository through dependency injection — it never knows whether it is talking to PostgreSQL, an in-memory dict, or a test double.

***

## Practical Example: Auth Service with Onion Structure

```python
# Domain ring: pure domain objects
# auth/domain/models.py

from dataclasses import dataclass
from enum import Enum

class Role(Enum):
    ADMIN = "admin"
    CASHIER = "cashier"
    MANAGER = "manager"

@dataclass(frozen=True)
class Email:
    value: str

    def __post_init__(self):
        if "@" not in self.value:
            raise ValueError(f"Invalid email: {self.value}")

@dataclass
class User:
    id: int | None
    tenant_id: str
    email: Email
    password_hash: str
    role: Role
    is_active: bool = True

    def can_access_tenant(self, tenant_id: str) -> bool:
        return self.tenant_id == tenant_id and self.is_active

    def has_role(self, required_role: Role) -> bool:
        return self.role == required_role
```

```python
# Application ring: use case
# auth/application/use_cases/authenticate.py

from ..repositories import UserRepository
from ..services import PasswordHasher, TokenService
from ...domain.models import User

class AuthenticateUserUseCase:
    def __init__(
        self,
        user_repo: UserRepository,
        hasher: PasswordHasher,    # Interface — not bcrypt directly
        token_service: TokenService # Interface — not PyJWT directly
    ):
        self._repo = user_repo
        self._hasher = hasher
        self._token_service = token_service

    def execute(self, tenant_id: str, email: str, password: str) -> str:
        user = self._repo.find_by_email(tenant_id, email)

        if user is None:
            raise ValueError("Invalid credentials")

        if not self._hasher.verify(password, user.password_hash):
            raise ValueError("Invalid credentials")

        if not user.can_access_tenant(tenant_id):
            raise PermissionError("Account inactive or wrong tenant")

        return self._token_service.generate(user)
```

```python
# UI ring: FastAPI router
# auth/ui/router.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from ..application.use_cases.authenticate import AuthenticateUserUseCase
from ..infrastructure.db.user_repository import SQLUserRepository
from ..infrastructure.security.bcrypt_hasher import BcryptPasswordHasher
from ..infrastructure.security.jwt_service import JWTTokenService
from .schemas import LoginRequest, TokenResponse

router = APIRouter()

@router.post("/login", response_model=TokenResponse)
def login(body: LoginRequest, db: Session = Depends(get_db)):
    use_case = AuthenticateUserUseCase(
        user_repo=SQLUserRepository(db),
        hasher=BcryptPasswordHasher(),
        token_service=JWTTokenService()
    )
    try:
        token = use_case.execute(body.tenant_id, body.email, body.password)
        return TokenResponse(access_token=token)
    except (ValueError, PermissionError) as e:
        raise HTTPException(status_code=401, detail="Invalid credentials")
```

The domain model `User` has no database imports. The use case `AuthenticateUserUseCase` has no SQLAlchemy or bcrypt imports. The router wires everything together and handles HTTP.

***

## Project Structure

```
auth-service/
├── domain/
│   ├── models.py          # Entities, value objects — zero external imports
│   └── events.py          # Domain events
│
├── application/
│   ├── repositories.py    # Abstract interfaces for persistence
│   ├── services.py        # Abstract interfaces for external capabilities
│   └── use_cases/
│       ├── authenticate.py
│       ├── register_user.py
│       └── refresh_token.py
│
├── infrastructure/
│   ├── db/
│   │   ├── orm_models.py       # SQLAlchemy ORM — infrastructure only
│   │   └── user_repository.py  # Implements application/repositories.py
│   └── security/
│       ├── bcrypt_hasher.py    # Implements PasswordHasher interface
│       └── jwt_service.py      # Implements TokenService interface
│
└── ui/
    ├── router.py
    └── schemas.py
```

***

## Testing Without a Database

The major payoff of onion architecture: domain and use case tests require no database:

```python
# tests/test_authenticate_use_case.py

from auth.application.use_cases.authenticate import AuthenticateUserUseCase
from auth.domain.models import User, Email, Role

# In-memory test doubles — no database, no bcrypt overhead
class FakeUserRepository:
    def __init__(self, users: list[User]):
        self._users = {(u.tenant_id, u.email.value): u for u in users}

    def find_by_email(self, tenant_id, email):
        return self._users.get((tenant_id, email))

    def save(self, user): return user

class FakeHasher:
    def verify(self, plain: str, hashed: str) -> bool:
        return plain == hashed  # Simplified for tests

class FakeTokenService:
    def generate(self, user: User) -> str:
        return f"token:{user.tenant_id}:{user.email.value}"

def test_authenticate_valid_user():
    user = User(
        id=1, tenant_id="restaurant_01",
        email=Email("chef@myrestaurant.com"),
        password_hash="secret123",
        role=Role.CASHIER
    )
    use_case = AuthenticateUserUseCase(
        FakeUserRepository([user]),
        FakeHasher(),
        FakeTokenService()
    )
    token = use_case.execute("restaurant_01", "chef@myrestaurant.com", "secret123")
    assert token == "token:restaurant_01:chef@myrestaurant.com"

def test_authenticate_wrong_password():
    user = User(
        id=1, tenant_id="restaurant_01",
        email=Email("chef@myrestaurant.com"),
        password_hash="secret123",
        role=Role.CASHIER
    )
    use_case = AuthenticateUserUseCase(
        FakeUserRepository([user]),
        FakeHasher(),
        FakeTokenService()
    )
    try:
        use_case.execute("restaurant_01", "chef@myrestaurant.com", "wrongpassword")
        assert False, "Should have raised"
    except ValueError:
        pass
```

These tests run in milliseconds with no database setup, no Docker, no test fixtures beyond Python objects.

***

## When Onion Architecture Pays Off

Onion architecture is worth the structure when:

* The domain logic is complex and changes frequently
* The application must support multiple interfaces (REST API, CLI, background jobs)
* Infrastructure choices may change (switching databases, adding a new external service)
* Comprehensive unit testing of business rules is a priority
* A team of developers needs to work in the codebase simultaneously with clear boundaries

It is overkill for:

* Simple CRUD applications with minimal business logic
* Small personal tools
* Proof-of-concept work where speed matters more than structure

***

## Lessons Learned

* **The centre is the most valuable part.** If the domain ring is rich and clean, the rest of the system is easier.
* **Abstract interfaces feel like ceremony until you need to swap an implementation.** When I replaced my email provider, I implemented one interface and changed one line of wiring.
* **Dependency injection is the mechanism that makes the onion possible.** Without it, you are importing concrete implementations and the inversion is broken.
* **Do not put application logic in the infrastructure ring.** I have seen repositories with conditional business rules inside them. That is domain/application logic that has leaked outward.