Python Designmönster för ren arkitektur

Bygg underhållbara Python-applikationer med SOLID-designmönster

Rent arkitektur har revolutionerat hur utvecklare bygger skalbara, underhållbara applikationer genom att betona separation av ansvar och beroendestyrning.

I Python kombineras dessa principer med språkets dynamiska natur för att skapa flexibla, testbara system som utvecklas med affärskrav utan att bli teknisk skuld.

vibrant tech conference hall

Förstå ren arkitektur i Python

Ren arkitektur, introducerad av Robert C. Martin (Uncle Bob), organiserar programvara i koncentriska lager där beroenden pekar inåt mot kärnbusinesslogik. Denna arkitekturmodell säkerställer att dina applikations kritiska affärsregler förblir oberoende av ramverk, databaser och externa tjänster.

Grundläggande filosofi

Den grundläggande principen är enkel men kraftfull: affärslogik ska inte bero av infrastruktur. Dina domänentiteter, användningsfall och affärsregler ska fungera oavsett om du använder PostgreSQL eller MongoDB, FastAPI eller Flask, AWS eller Azure.

I Python passar denna filosofi perfekt med språkets “duck typing” och protokollorienterad programmering, vilket möjliggör ren separation utan den ceremonin som krävs i statiskt typade språk.

De fyra lagren i ren arkitektur

Entitetslager (Domän): Rena affärsobjekt med företagsövergripande affärsregler. Dessa är POJOs (Plain Old Python Objects) utan externa beroenden.

Användningsfallslager (Applikation): Applikationsspecifika affärsregler som orkestrerar datatillflödet mellan entiteter och externa tjänster.

Gränssnittsanpassningslager: Konverterar data mellan det format som är mest lämpligt för användningsfall och entiteter, och det format som krävs av externa enheter.

Ramverk & drivrutiner: Alla externa detaljer som databaser, webramverk och externa API:er.

SOLID-principer i Python

SOLID-principerna utgör grunden för ren arkitektur. Låt oss utforska hur varje princip manifesteras i Python. För en omfattande översikt över designmönster i Python, se Python Design Patterns Guide.

Single Responsibility Principle (SRP)

Varje klass ska ha en anledning att ändras:

# Dåligt: Flera ansvar
class UserManager:
    def create_user(self, user_data):
        # Skapa användare
        pass

    def send_welcome_email(self, user):
        # Skicka e-post
        pass

    def log_creation(self, user):
        # Logga till fil
        pass

# Bra: Separerade ansvar
class UserService:
    def __init__(self, repository, email_service, logger):
        self.repository = repository
        self.email_service = email_service
        self.logger = logger

    def create_user(self, user_data):
        user = User(**user_data)
        self.repository.save(user)
        self.email_service.send_welcome(user)
        self.logger.info(f"Användare skapad: {user.id}")
        return user

Open/Closed Principle (OCP)

Programentiteter ska vara öppna för utökning men stängda för modifiering:

from abc import ABC, abstractmethod
from typing import Protocol

# Användning av Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # Kreditkortlogik
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # PayPal-logik
        return True

# Lätt att utöka utan att modifiera befintlig kod
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Kryptovalutalogik
        return True

Liskov Substitution Principle (LSP)

Objekt ska kunna ersättas med sina undertyper utan att programmet bryts:

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def save(self, key: str, value: str) -> None:
        pass

    @abstractmethod
    def get(self, key: str) -> str:
        pass

class PostgreSQLStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # PostgreSQL-implementering
        pass

    def get(self, key: str) -> str:
        # PostgreSQL-implementering
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Redis-implementering
        pass

    def get(self, key: str) -> str:
        # Redis-implementering
        return ""

# Båda kan användas utbytbart
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

Interface Segregation Principle (ISP)

Klienter ska inte tvingas bero på gränssnitt de inte använder:

# Dåligt: Fett gränssnitt
class Worker(ABC):
    @abstractmethod
    def work(self): pass

    @abstractmethod
    def eat(self): pass

    @abstractmethod
    def sleep(self): pass

# Bra: Segregerade gränssnitt
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("Arbetar")

    def eat(self) -> None:
        print("Äter")

class Robot:
    def work(self) -> None:
        print("Arbetar")
    # Inget ätmetod behövs

Dependency Inversion Principle (DIP)

Högre nivåmoduler ska inte bero på lågnivåmoduler. Båda ska bero på abstraktioner:

from typing import Protocol

# Abstraktion
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# Lågnivåmodul
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # SMTP-implementering
        pass

# Högre nivåmodul beroende av abstraktion
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender

    def register(self, email: str, name: str):
        # Registreringslogik
        self.email_sender.send(
            to=email,
            subject="Välkommen!",
            body=f"Hej {name}"
        )

Repository-mönster: Abstrahering av dataåtkomst

Repository-mönstret ger ett samlingsliknande gränssnitt för att komma åt domänobjekt, döljer detaljerna om datalagring.

Grundläggande repository-implementering

from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID
    email: str
    name: str
    is_active: bool = True

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass

    @abstractmethod
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]:
        pass

    @abstractmethod
    def list_all(self) -> List[User]:
        pass

    @abstractmethod
    def delete(self, user_id: UUID) -> bool:
        pass

SQLAlchemy-implementering

from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'

    id = Column(PGUUID(as_uuid=True), primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session

    def save(self, user: User) -> User:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
        self.session.add(user_model)
        self.session.commit()
        return user

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def get_by_email(self, email: str) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.email == email
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def list_all(self) -> List[User]:
        users = self.session.query(UserModel).all()
        return [
            User(
                id=u.id,
                email=u.email,
                name=u.name,
                is_active=u.is_active
            )
            for u in users
        ]

    def delete(self, user_id: UUID) -> bool:
        result = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).delete()
        self.session.commit()
        return result > 0

Minnesintern repository för testning

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users: dict[UUID, User] = {}

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

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        return self.users.get(user_id)

    def get_by_email(self, email: str) -> Optional[User]:
        for user in self.users.values():
            if user.email == email:
                return user
        return None

    def list_all(self) -> List[User]:
        return list(self.users.values())

    def delete(self, user_id: UUID) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

Service Layer: Orchestrerar Affärslogik

Service Layer implementerar användningsfall och orkestrerar flödet mellan lagringsplatser, externa tjänster och domänlogik.

from typing import Optional
from uuid import uuid4

class UserAlreadyExistsError(Exception):
    pass

class UserNotFoundError(Exception):
    pass

class UserService:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailSender,
        event_publisher: 'EventPublisher'
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.event_publisher = event_publisher

    def register_user(self, email: str, name: str) -> User:
        # Kontrollera om användare finns
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"Användare med e-post {email} finns redan")

        # Skapa ny användare
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )

        # Spara i lagring
        user = self.user_repository.save(user)

        # Skicka välkomstmail
        self.email_service.send(
            to=user.email,
            subject="Välkommen!",
            body=f"Hej {user.name}, välkommen till vår plattform!"
        )

        # Publicera händelse
        self.event_publisher.publish('user.registered', {
            'user_id': str(user.id),
            'email': user.email
        })

        return user

    def deactivate_user(self, user_id: UUID) -> User:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"Användare {user_id} hittades inte")

        user.is_active = False
        user = self.user_repository.save(user)

        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })

        return user

Beroendeinjektion i Python

Pythons dynamiska natur gör beroendeinjektion enkelt utan att kräva tunga ramverk.

Konstruktorinjektion

class OrderService:
    def __init__(
        self,
        order_repository: 'OrderRepository',
        payment_processor: PaymentProcessor,
        notification_service: 'NotificationService'
    ):
        self.order_repository = order_repository
        self.payment_processor = payment_processor
        self.notification_service = notification_service

    def place_order(self, order_data: dict):
        # Använd injicerade beroenden
        pass

Enkel beroendekontainer

from typing import Dict, Type, Callable, Any

class Container:
    def __init__(self):
        self._services: Dict[Type, Callable] = {}
        self._singletons: Dict[Type, Any] = {}

    def register(self, interface: Type, factory: Callable):
        self._services[interface] = factory

    def register_singleton(self, interface: Type, instance: Any):
        self._singletons[interface] = instance

    def resolve(self, interface: Type):
        if interface in self._singletons:
            return self._singletons[interface]

        factory = self._services.get(interface)
        if factory:
            return factory(self)

        raise ValueError(f"Ingen registrering hittad för {interface}")

# Användning
def create_container() -> Container:
    container = Container()

    # Registrera tjänster
    container.register_singleton(
        Session,
        sessionmaker(bind=create_engine('postgresql://...'))()
    )

    container.register(
        UserRepository,
        lambda c: SQLAlchemyUserRepository(c.resolve(Session))
    )

    container.register(
        EmailSender,
        lambda c: SMTPEmailSender()
    )

    container.register(
        UserService,
        lambda c: UserService(
            c.resolve(UserRepository),
            c.resolve(EmailSender),
            c.resolve(EventPublisher)
        )
    )

    return container

Hexagonal Arkitektur (Portar och Anpassare)

Hexagonal Arkitektur placerar affärslogik i centrum med anpassare som hanterar extern kommunikation.

Definiera Portar (Gränssnitt)

# Input Port (Primär)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Output Port (Sekundär)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...

    def find_by_email(self, email: str) -> Optional[User]:
        ...

Implementera Anpassare

from pydantic import BaseModel, EmailStr

# Input Anpassare (REST API)
from fastapi import FastAPI, Depends, HTTPException

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str

class CreateUserResponse(BaseModel):
    id: str
    email: str
    name: str

app = FastAPI()

@app.post("/users", response_model=CreateUserResponse)
def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(get_user_service)
):
    try:
        user = user_service.register_user(
            email=request.email,
            name=request.name
        )
        return CreateUserResponse(
            id=str(user.id),
            email=user.email,
            name=user.name
        )
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=400, detail=str(e))

# Output Anpassare (Databas)
# Redan implementerad som SQLAlchemyUserRepository

Domänorienterade Designmönster

Värdeobjekt

Oföränderliga objekt definierade av sina attribut:

from dataclasses import dataclass
from typing import Pattern
import re

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

    EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')

    def __post_init__(self):
        if not self.EMAIL_PATTERN.match(self.value):
            raise ValueError(f"Ogiltig e-post: {self.value}")

    def __str__(self):
        return self.value

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Belopp kan inte vara negativt")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Ostödd valuta: {self.currency}")

    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Kan inte lägga till olika valutor")
        return Money(self.amount + other.amount, self.currency)

Aggregat

Kluster av domänobjekt behandlade som en enda enhet:

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class OrderItem:
    product_id: UUID
    quantity: int
    price: Money

    def total(self) -> Money:
        return Money(
            self.price.amount * self.quantity,
            self.price.currency
        )

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    items: List[OrderItem] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)

    def add_item(self, product_id: UUID, quantity: int, price: Money):
        item = OrderItem(product_id, quantity, price)
        self.items.append(item)

    def remove_item(self, product_id: UUID):
        self.items = [
            item for item in self.items
            if item.product_id != product_id
        ]

    def total(self) -> Money:
        if not self.items:
            return Money(0, "USD")

        return sum(
            (item.total() for item in self.items),
            Money(0, self.items[0].price.currency)
        )

    def confirm(self):
        if not self.items:
            raise ValueError("Kan inte bekräfta tom order")
        if self.status != "pending":
            raise ValueError("Order redan bearbetad")
        self.status = "confirmed"

Domänhändelser

Domänhändelser möjliggör lös koppling mellan komponenter och stöder händelsedrivna arkitekturer. För produktionsskaliga händelsedrivna system, överväg att implementera händelseflöden med tjänster som AWS Kinesis - se Bygga händelsedrivna mikrotjänster med AWS Kinesis för en detaljerad guide.

from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.now)

@dataclass
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total: Money

class EventPublisher:
    def __init__(self):
        self._handlers: Dict[Type, List[Callable]] = {}

    def subscribe(self, event_type: Type, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def publish(self, event: DomainEvent):
        event_type = type(event)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            handler(event)

Moderna Python-funktioner för ren arkitektur

Python’s moderna funktioner gör det mer elegant och typ-säkert att implementera ren arkitektur. Om du behöver en snabb referens för Python-syntax och funktioner, kolla in Python Cheatsheet.

Typhjälpmedel och Protokoll

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict:
        ...

    @classmethod
    def from_dict(cls, data: dict) -> 'Serializable':
        ...

def serialize(obj: Serializable) -> dict:
    return obj.to_dict()

Pydantic för validering

from pydantic import BaseModel, Field, validator
from typing import Optional

class CreateUserDTO(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('name')
    def name_must_not_contain_numbers(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Name cannot contain numbers')
        return v

    class Config:
        frozen = True  # Make immutable

Async/Await för I/O-operationer

Python’s async/await-syntax är särskilt kraftfull för I/O-bundna operationer i ren arkitektur, vilket tillåter icke-blockerande interaktioner med databaser och externa tjänster. När du distribuerar Python-applikationer till serverlösa plattformar blir det viktigt att förstå prestandaegenskaper - se AWS lambda performance: JavaScript vs Python vs Golang för insikter om hur du optimerar Python-serverlösa funktioner.

from typing import List
import asyncio

class AsyncUserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass

    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

class AsyncUserService:
    def __init__(self, repository: AsyncUserRepository):
        self.repository = repository

    async def register_user(self, email: str, name: str) -> User:
        user = User(id=uuid4(), email=email, name=name)
        return await self.repository.save(user)

    async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
        tasks = [self.repository.get_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

Projektstruktur - bästa praxis

Rätt projektorganisation är avgörande för att upprätthålla ren arkitektur. Innan du sätter upp din projektstruktur, se till att du använder Python-virtualmiljöer för beroendeisolering. venv Cheatsheet täcker allt du behöver veta om att hantera virtualmiljöer. För moderna Python-projekt, överväg att använda uv - New Python Package, Project, and Environment Manager, som erbjuder snabbare pakethantering och projektinställning.

my_application/
├── domain/                 # Företagsregler
│   ├── __init__.py
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   └── money.py
│   ├── events/
│   │   ├── __init__.py
│   │   └── user_events.py
│   └── exceptions.py
│
├── application/            # Applikationsregler
│   ├── __init__.py
│   ├── use_cases/
│   │   ├── __init__.py
│   │   ├── create_user.py
│   │   └── place_order.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── ports/
│       ├── __init__.py
│       ├── repositories.py
│       └── external_services.py
│
├── infrastructure/         # Externa gränssnitt
│   ├── __init__.py
│   ├── persistence/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── models.py
│   │   │   └── repositories.py
│   │   └── mongodb/
│   │       └── repositories.py
│   ├── messaging/
│   │   ├── __init__.py
│   │   └── rabbitmq_publisher.py
│   ├── external_services/
│   │   ├── __init__.py
│   │   └── email_service.py
│   └── config.py
│
├── presentation/           # UI/API-lager
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── dependencies.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── orders.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user_schemas.py
│   └── cli/
│       └── commands.py
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── main.py                 # Applikationens ingångspunkt
├── container.py            # Beroendeinjektion
├── pyproject.toml
└── README.md

Testning av ren arkitektur

Enhetstestning av domänlogik

import pytest
from uuid import uuid4

def test_user_creation():
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert user.email == "test@example.com"
    assert user.is_active is True

def test_order_total_calculation():
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    order.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert order.total().amount == 25.0

Integrationstestning med repository

@pytest.fixture
def in_memory_repository():
    return InMemoryUserRepository()

def test_user_repository_save_and_retrieve(in_memory_repository):
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )

    saved_user = in_memory_repository.save(user)
    retrieved_user = in_memory_repository.get_by_id(user.id)

    assert retrieved_user is not None
    assert retrieved_user.email == user.email

Testning av servicelager

from unittest.mock import Mock

def test_user_registration():
    # Arrange
    mock_repository = Mock(spec=UserRepository)
    mock_repository.get_by_email.return_value = None
    mock_repository.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )

    mock_email = Mock(spec=EmailSender)
    mock_events = Mock(spec=EventPublisher)

    service = UserService(mock_repository, mock_email, mock_events)

    # Act
    user = service.register_user("test@example.com", "Test")

    # Assert
    assert user.email == "test@example.com"
    mock_repository.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_events.publish.assert_called_once()

Vanliga fallgropar och hur man undviker dem

Överdimensionering

Implementera inte ren arkitektur för enkla CRUD-applikationer. Börja enkelt och refaktorera när komplexiteten ökar.

Läckande abstraktioner

Se till att domänentiteter inte innehåller databasannotationer eller ramverksspecifik kod:

# Dåligt
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Ramverk läcker in i domänen

# Bra
@dataclass
class User:
    id: UUID  # Ren domänobjekt

Cirkulära beroenden

Använd beroendeinjektion och gränssnitt för att bryta cirkulära beroenden mellan lagren.

Att ignorera kontext

Ren arkitektur är inte en passform för alla. Anpassa lagrets strikthet baserat på projektets storlek och teamets expertis.

Användbara länkar