Skip to main content

​
Introduction to Clean Architecture

Factus API implements clean architecture (also known as hexagonal architecture or ports and adapters pattern). This architectural style ensures that business logic is independent of external frameworks, databases, and APIs.
Clean Architecture Layers

​
Core Principles

Dependency Rule

Dependencies point inward. Domain layer never depends on infrastructure.

Domain Isolation

Business logic has zero framework dependencies - pure Python only.

Interface Segregation

Components communicate through abstract interfaces (ports).

Testability

Easy to test business logic without external dependencies.

​
The Layers

​
Domain Layer (Inner Circle)

The domain layer contains business entities and defines contracts (interfaces) that outer layers must implement. It has no dependencies on external libraries. 
# app/src/domain/models/auth_token.py
from pydantic import BaseModel

class AuthToken(BaseModel):
    """Pure domain model - no framework dependencies"""
    access_token: str
    token_type: str
    expires_in: int
    refresh_token: str | None = None
Key Insight: The domain layer defines what operations are needed (IAuthGateway) but not how they’re implemented. The implementation details live in the infrastructure layer.

​
Infrastructure Layer (Outer Circle)

The infrastructure layer implements the interfaces (ports) defined by the domain layer. This is where we interact with external APIs, databases, and frameworks.
Adapter (Implementation)
# app/src/infrastructure/gateways/factus_auth_gateway.py
import httpx
from app.src.domain.interfaces.auth_gateway import IAuthGateway
from app.src.domain.models.auth_token import AuthToken
from app.src.core.config import settings

class FactusAuthGateway(IAuthGateway):
    """Adapter: implements the IAuthGateway port using httpx"""
    
    def __init__(self, base_url: str, client_id: str, client_secret: str):
        self.base_url = base_url.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret

    async def authenticate(self, email: str, password: str) -> AuthToken:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/oauth/token",
                data={
                    "grant_type": "password",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "username": email,
                    "password": password,
                },
                headers={"Accept": "application/json"}
            )

        if not response.is_success:
            raise Exception(self._parse_error(response, "Error de autenticación con Factus"))

        data = response.json()
        return AuthToken(
            access_token=data["access_token"],
            token_type=data["token_type"],
            expires_in=data["expires_in"],
            refresh_token=data.get("refresh_token")
        )

    async def refresh_token(self, refresh_token: str) -> AuthToken:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/oauth/token",
                data={
                    "grant_type": "refresh_token",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "refresh_token": refresh_token,
                },
                headers={"Accept": "application/json"}
            )

        if not response.is_success:
            raise Exception(self._parse_error(response, "Error al refrescar el token de Factus"))

        data = response.json()
        return AuthToken(
            access_token=data["access_token"],
            token_type=data["token_type"],
            expires_in=data["expires_in"],
            refresh_token=data.get("refresh_token")
        )

    def _parse_error(self, response: httpx.Response, default: str) -> str:
        try:
            error_data = response.json()
            return (
                error_data.get("message")
                or error_data.get("error_description")
                or error_data.get("error")
                or default
            )
        except Exception:
            return response.text or f"HTTP {response.status_code}"

​
The Gateway Pattern

Gateways are the primary pattern for integrating with external services. Each gateway:
  1. Implements a domain interface (e.g., IAuthGateway, IInvoiceGateway)
  2. Encapsulates HTTP communication using httpx.AsyncClient
  3. Transforms external data into domain models
  4. Handles errors and provides meaningful messages

​
Invoice Gateway Example

Interface Definition
# app/src/domain/interfaces/invoice_gateway.py
from abc import ABC, abstractmethod
from app.src.domain.models.invoice import Invoice, InvoiceResponse

class IInvoiceGateway(ABC):
    @abstractmethod
    async def create_invoice(self, invoice: Invoice, token: str) -> InvoiceResponse:
        """Sends an invoice to the external API (Factus) and returns the validation response."""
        pass

    @abstractmethod
    async def download_pdf(self, number: str, token: str) -> DownloadResponse:
        """Downloads the PDF of an invoice."""
        pass

    @abstractmethod
    async def download_xml(self, number: str, token: str) -> DownloadResponse:
        """Downloads the XML of an invoice."""
        pass
By defining interfaces in the domain layer, we can:
  • Mock gateways for testing
  • Swap implementations without changing business logic
  • Support multiple providers (e.g., different e-invoicing systems)

​
Dependency Injection with FastAPI

FastAPI’s dependency injection system connects the layers together. We define factory functions that create gateway instances:
Dependency Configuration
# app/src/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from app.src.core.config import settings
from app.src.domain.models.user import TokenData, User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")

# Gateway factories
def get_auth_gateway() -> FactusAuthGateway:
    """Factory for creating auth gateway instances"""
    return FactusAuthGateway(
        base_url=settings.FACTUS_BASE_URL,
        client_id=settings.FACTUS_CLIENT_ID,
        client_secret=settings.FACTUS_CLIENT_SECRET
    )

def get_invoice_gateway() -> FactusInvoiceGateway:
    """Factory for creating invoice gateway instances"""
    return FactusInvoiceGateway(
        base_url=settings.FACTUS_BASE_URL
    )

# Authentication dependency
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    
    # In a real implementation, this would query a database
    user_dict = fake_users_db.get(token_data.username)
    if user_dict is None:
        raise credentials_exception
    return User(**user_dict)

​
Using Dependencies in Endpoints

Route Handler
# app/src/api/v1/endpoints/auth.py
from fastapi import APIRouter, Depends, HTTPException
from app.src.api.deps import get_auth_gateway
from app.src.domain.interfaces.auth_gateway import IAuthGateway
from app.src.api.v1.schemas.auth import LoginRequest, LoginResponse

router = APIRouter()

@router.post("/login", response_model=LoginResponse)
async def login(
    request: LoginRequest,
    gateway: IAuthGateway = Depends(get_auth_gateway)
):
    """Authenticate with Factus and return an access token."""
    try:
        token = await gateway.authenticate(request.email, request.password)
        return LoginResponse(
            access_token=token.access_token,
            token_type=token.token_type,
            expires_in=token.expires_in
        )
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))
Notice how the endpoint receives an IAuthGateway interface, not the concrete FactusAuthGateway class. This makes the code testable and flexible.

​
Benefits of This Architecture

Mock gateways and test business logic in isolation:
class MockAuthGateway(IAuthGateway):
    async def authenticate(self, email: str, password: str) -> AuthToken:
        return AuthToken(
            access_token="mock_token",
            token_type="Bearer",
            expires_in=3600
        )

# Test without hitting the real Factus API
async def test_login():
    mock_gateway = MockAuthGateway()
    result = await login_handler(request, gateway=mock_gateway)
    assert result.access_token == "mock_token"
Swap implementations without changing business logic:
  • Switch from Factus to another e-invoicing provider
  • Add caching layer by wrapping gateways
  • Support multiple providers simultaneously
Clear separation of concerns:
  • Domain logic in domain/
  • External integrations in infrastructure/
  • HTTP handling in api/
  • Each layer can evolve independently
Business logic doesn’t depend on FastAPI:
  • Domain models are pure Python/Pydantic
  • Can migrate to Django, Flask, or CLI without rewriting logic
  • Domain layer is framework-agnostic

​
Best Practices

1

Define interfaces in domain layer

Always create abstract base classes for external dependencies:
class IPaymentGateway(ABC):
    @abstractmethod
    async def process_payment(self, amount: Decimal) -> PaymentResult:
        pass
2

Keep domain models pure

Domain models should never import from infrastructure/ or api/:
# ✅ Good: Pure Pydantic model
class Invoice(BaseModel):
    number: str
    amount: Decimal

# ❌ Bad: Importing from infrastructure
from app.src.infrastructure.db import Base
class Invoice(Base):
    ...
3

Implement gateways in infrastructure

All external API communication belongs in infrastructure/gateways/:
class FactusInvoiceGateway(IInvoiceGateway):
    async def create_invoice(self, invoice: Invoice, token: str) -> InvoiceResponse:
        async with httpx.AsyncClient() as client:
            response = await client.post(...)
            return InvoiceResponse(**response.json())
4

Use dependency injection

Connect layers using FastAPI’s Depends():
@router.post("/invoice")
async def create_invoice(
    invoice: InvoiceCreate,
    gateway: IInvoiceGateway = Depends(get_invoice_gateway)
):
    return await gateway.create_invoice(invoice)

Architecture Overview

Learn about the overall project structure

Configuration

Understand settings and environment variables

Quickstart Guide

Set up your development environment

Authentication API

Explore the complete API documentation

Build docs developers (and LLMs) love

Get started for freeTalk to us