Hexagonal Architecture (Python)

# domain.py
from abc import ABC, abstractmethod
from pydantic import BaseModel

# Domain Model (Could be dataclass, attrs class, Pydantic model or simple class)
class User(BaseModel):
    id: int | None = None
    email: str
    username: str

# Outbound Port: What the core expects from a database 
# (interfaces that would be implemented via ORM, SQL or other depending on the adapter)
class UserRepositoryPort(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    def get_by_email(self, email: str) -> User | None:
        pass

# Core Use Case / Service
class UserService:
    def __init__(self, user_repo: UserRepositoryPort):
        self.user_repo = user_repo  # Injecting the port

    def register_user(self, email: str, username: str) -> User:
        if self.user_repo.get_by_email(email):
            raise ValueError("Email already exists")
        
        new_user = User(email=email, username=username)
        return self.user_repo.save(new_user)

# Implementation using SQLAlchemy 2.0 style
# postgres_adapter.py
from sqlalchemy import String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from domain import UserRepositoryPort, User

class Base(DeclarativeBase):
    pass

# Database table model using modern Mapped types
class UserModel(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    username: Mapped[str] = mapped_column(String)

class PostgresUserRepository(UserRepositoryPort):
    def __init__(self, db_session: Session):
        self.db_session = db_session

    def save(self, user: User) -> User:
        db_user = UserModel(email=user.email, username=user.username)
        self.db_session.add(db_user)
        self.db_session.commit()
        self.db_session.refresh(db_user)
        return User(id=db_user.id, email=db_user.email, username=db_user.username)

    def get_by_email(self, email: str) -> User | None:
        stmt = select(UserModel).where(UserModel.email == email)
        db_user = self.db_session.scalars(stmt).first()
        
        if not db_user:
            return None
        return User(id=db_user.id, email=db_user.email, username=db_user.username)

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from domain import UserService
from postgres_adapter import PostgresUserRepository, Base
from pydantic import BaseModel

# Standard SQLAlchemy setup, on production or other env it might require permissions
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency Injection helper
def get_user_service():
    db = SessionLocal()
    try:
        # Wire adapter into port, and port into service
        repo = PostgresUserRepository(db)
        yield UserService(repo)
    finally:
        db.close()

# FastAPI Request Schema
class UserCreateRequest(BaseModel):
    email: str
    username: str

@app.post("/users/", response_model=dict, status_code=status.HTTP_201_CREATED)
def create_user(request: UserCreateRequest, service: UserService = Depends(get_user_service)):
    try:
        user = service.register_user(email=request.email, username=request.username)
        return {"id": user.id, "email": user.email, "username": user.username}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

# test_user_service.py service mock helper
import pytest
from domain import UserRepositoryPort, User, UserService

# An Outbound Adapter created strictly for testing purposes
class InMemoryUserRepository(UserRepositoryPort):
    def __init__(self):
        self.users: dict[str, User] = {}
        self._id_counter = 1

    def save(self, user: User) -> User:
        user.id = self._id_counter
        self.users[user.email] = user
        self._id_counter += 1
        return user

    def get_by_email(self, email: str) -> User | None:
        return self.users.get(email)

def test_register_user_successfully():
    # Arrange: Setup the mock repository and the service
    mock_repo = InMemoryUserRepository()
    service = UserService(mock_repo)

    # Act: Run the production logic
    new_user = service.register_user(email="[email protected]", username="testuser")

    # Assert: Verify the results
    assert new_user.id == 1
    assert new_user.email == "[email protected]"
    # Ensure it actually saved into our fake repository
    assert mock_repo.get_by_email("[email protected]") is not None


def test_register_user_duplicate_email_raises_error():
    # Arrange: Setup repository and pre-populate an existing user
    mock_repo = InMemoryUserRepository()
    service = UserService(mock_repo)
    service.register_user(email="[email protected]", username="first_user")

    # Act & Assert: Verify that duplicating an email raises a ValueError
    with pytest.raises(ValueError) as exc_info:
        service.register_user(email="[email protected]", username="second_user")
        
    assert str(exc_info.value) == "Email already exists"

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

# --- 1. SETUP DATABASE ---
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# --- 2. FASTAPI DEPENDENCY HELPERS ---
# This is your single source of truth for managing the DB session lifecycle
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Factory for UserService
def get_user_service(db: Session = Depends(get_db)) -> UserService:
    # 1. Instantiate the DB adapter
    user_repo = PostgresUserRepository(db)
    # 2. Inject it into the service and return it
    return UserService(user_repo)

# Factory for OrderService (Sharing the exact same DB session helper!)
def get_order_service(db: Session = Depends(get_db)) -> OrderService:
    order_repo = PostgresOrderRepository(db)
    return OrderService(order_repo)

app = FastAPI()

# Endpoint 1: Uses the User Service
@app.post("/users/")
def create_user(request: UserCreate, service: UserService = Depends(get_user_service)):
    return service.register_user(request.email, request.username)

# Endpoint 2: Uses the User Service (Reusing the same service structure!)
@app.get("/users/{user_id}")
def get_user(user_id: int, service: UserService = Depends(get_user_service)):
    return service.get_user_profile(user_id)

# Endpoint 3: Uses a completely different Service, but uses the same DB engine under the hood
@app.post("/orders/")
def create_order(request: OrderCreate, service: OrderService = Depends(get_order_service)):
    return service.place_order(request.user_id, request.items)

# domain/ports.py
from abc import ABC, abstractmethod
from typing import Self
from domain.models import User, Order  # Assuming simple Pydantic/dataclass models

class UserRepositoryPort(ABC):
    @abstractmethod
    def save(self, user: User) -> User: pass
    
    @abstractmethod
    def get_by_id(self, user_id: int) -> User | None: pass

class OrderRepositoryPort(ABC):
    @abstractmethod
    def save(self, order: Order) -> Order: pass


# The Unit of Work Port
class UnitOfWorkPort(ABC):
    users: UserRepositoryPort
    orders: OrderRepositoryPort

    @abstractmethod
    def __enter__(self) -> Self: pass

    @abstractmethod
    def __exit__(self, exc_type, exc_val, exc_tb): pass

    @abstractmethod
    def commit(self): pass

    @abstractmethod
    def rollback(self): pass

# domain/services.py
from domain.ports import UnitOfWorkPort
from domain.models import Order

class OrderPlacementService:
    def __init__(self, uow: UnitOfWorkPort):
        self.uow = uow

    def place_order(self, user_id: int, item_name: str, price: float) -> Order:
        # We wrap our database operations in the UoW context manager
        with self.uow:
            # 1. Use the user repository attached to the UoW
            user = self.uow.users.get_by_id(user_id)
            if not user:
                raise ValueError("User not found")
            
            # 2. Business Rule validation
            if user.is_suspended:
                raise ValueError("Suspended users cannot place orders")
                
            # 3. Create and save order using the order repository attached to the UoW
            new_order = Order(user_id=user_id, item_name=item_name, price=price)
            saved_order = self.uow.orders.save(new_order)
            
            # 4. Explicitly commit if all operations pass successfully
            self.uow.commit()
            return saved_order
            
        # If any exception is raised within the `with` block, 
        # the UoW's __exit__ method automatically triggers a rollback.

# adapters/postgres.py
from sqlalchemy.orm import Session
from domain.ports import UnitOfWorkPort
from adapters.repositories import PostgresUserRepository, PostgresOrderRepository

class SQLAlchemyUnitOfWork(UnitOfWorkPort):
    def __init__(self, session_factory):
        self.session_factory = session_factory  # A sessionmaker instance

    def __enter__(self):
        self.session: Session = self.session_factory()
        # Bind repositories to the newly opened session
        self.users = PostgresUserRepository(self.session)
        self.orders = PostgresOrderRepository(self.session)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            self.rollback()  # An error occurred inside the 'with' block
        self.session.close()

    def commit(self):
        self.session.commit()

    def rollback(self):
        self.session.rollback()

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from adapters.postgres import SQLAlchemyUnitOfWork
from domain.services import OrderPlacementService
from pydantic import BaseModel

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

# Single DI Factory for the use case
def get_order_service() -> OrderPlacementService:
    uow = SQLAlchemyUnitOfWork(SessionLocal)
    return OrderPlacementService(uow)

class OrderRequest(BaseModel):
    user_id: int
    item_name: str
    price: float

@app.post("/orders/", status_code=status.HTTP_201_CREATED)
def create_order(payload: OrderRequest, service: OrderPlacementService = Depends(get_order_service)):
    try:
        order = service.place_order(
            user_id=payload.user_id, 
            item_name=payload.item_name, 
            price=payload.price
        )
        return {"status": "success", "order_id": order.id}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))