python-patterns

Python-Entwicklungsmuster

Idiomatische Python-Muster und Best Practices für die Entwicklung robuster, effizienter und wartbarer Anwendungen.

Wann aktivieren

• Schreiben von neuem Python-Code
• Überprüfen von Python-Code
• Refactoring von vorhandenem Python-Code
• Entwerfen von Python-Paketen/-Modulen

Kernprinzipien

1. Lesbarkeit zählt

Python priorisiert Lesbarkeit. Code sollte offensichtlich und leicht verständlich sein.

# Gut: klar und lesbar
def get_active_users(users: list[User]) -> list[User]:
    """Gibt nur aktive Benutzer aus der bereitgestellten Liste zurück."""
    return [user for user in users if user.is_active]


# Schlecht: clever, aber verwirrend
def get_active_users(u):
    return [x for x in u if x.a]

2. Explizit ist besser als implizit

Vermeiden Sie Magie; seien Sie klar darüber, was Ihr Code tut.

# Gut: explizite Konfiguration
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Schlecht: versteckte Seiteneffekte
import some_module
some_module.setup()  # Was macht das?

3. EAFP – Es ist einfacher um Vergebung zu bitten als um Erlaubnis

Python bevorzugt die Ausnahmebehandlung gegenüber der Überprüfung von Bedingungen.

# Gut: EAFP-Stil
def get_value(dictionary: dict, key: str, default_value: Any = None) -> Any:
    try:
        return dictionary[key]
    except KeyError:
        return default_value

# Schlecht: LBYL (Look Before You Leap)-Stil
def get_value(dictionary: dict, key: str, default_value: Any = None) -> Any:
    if key in dictionary:
        return dictionary[key]
    else:
        return default_value

Typannotationen

Grundlegende Typannotationen

from typing import Optional, List, Dict, Any

def process_user(
    user_id: str,
    data: Dict[str, Any],
    active: bool = True
) -> Optional[User]:
    """Verarbeitet einen Benutzer und gibt den aktualisierten User oder None zurück."""
    if not active:
        return None
    return User(user_id, data)

Moderne Typannotationen (Python 3.9+)

# Python 3.9+ – Verwenden Sie eingebaute Typen
def process_items(items: list[str]) -> dict[str, int]:
    return {item: len(item) for item in items}

# Python 3.8 und früher – Verwenden Sie das typing-Modul
from typing import List, Dict

def process_items(items: List[str]) -> Dict[str, int]:
    return {item: len(item) for item in items}

Typ-Aliase und TypeVar

from typing import TypeVar, Union

# Typ-Alias für komplexe Typen
JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None]

def parse_json(data: str) -> JSON:
    return json.loads(data)

# Generische Typen
T = TypeVar('T')

def first(items: list[T]) -> T | None:
    """Gibt das erste Element oder None zurück, wenn die Liste leer ist."""
    return items[0] if items else None

Duck-Typing mit Protocol

from typing import Protocol

class Renderable(Protocol):
    def render(self) -> str:
        """Rendert das Objekt in einen String."""

def render_all(items: list[Renderable]) -> str:
    """Rendert alle Elemente, die das Renderable-Protocol implementieren."""
    return "\n".join(item.render() for item in items)

Fehlerbehandlungsmuster

Spezifische Ausnahmebehandlung

# Gut: Spezifische Ausnahmen abfangen
def load_config(path: str) -> Config:
    try:
        with open(path) as f:
            return Config.from_json(f.read())
    except FileNotFoundError as e:
        raise ConfigError(f"Konfigurationsdatei nicht gefunden: {path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"Ungültiges JSON in der Konfiguration: {path}") from e

# Schlecht: Bloßes except
def load_config(path: str) -> Config:
    try:
        with open(path) as f:
            return Config.from_json(f.read())
    except:
        return None  # Stiller Fehlschlag!

Ausnahmeverkettung

def process_data(data: str) -> Result:
    try:
        parsed = json.loads(data)
    except json.JSONDecodeError as e:
        # Ausnahmen verketten, um den Traceback zu erhalten
        raise ValueError(f"Fehler beim Parsen der Daten: {data}") from e

Benutzerdefinierte Ausnahmehierarchie

class AppError(Exception):
    """Basisausnahme für alle Anwendungsfehler."""
    pass

class ValidationError(AppError):
    """Wird ausgelöst, wenn die Eingabevalidierung fehlschlägt."""
    pass

class NotFoundError(AppError):
    """Wird ausgelöst, wenn eine angeforderte Ressource nicht gefunden wird."""
    pass

# Verwendung
def get_user(user_id: str) -> User:
    user = db.find_user(user_id)
    if not user:
        raise NotFoundError(f"Benutzer nicht gefunden: {user_id}")
    return user

Kontextmanager

Ressourcenverwaltung

# Gut: Verwendung von Kontextmanagern
def process_file(path: str) -> str:
    with open(path, 'r') as f:
        return f.read()

# Schlecht: Manuelle Ressourcenverwaltung
def process_file(path: str) -> str:
    f = open(path, 'r')
    try:
        return f.read()
    finally:
        f.close()

Benutzerdefinierte Kontextmanager

from contextlib import contextmanager

@contextmanager
def timer(name: str):
    """Kontextmanager zum Zeitmessen eines Code-Blocks."""
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"{name} dauerte {elapsed:.4f} Sekunden")

# Verwendung
with timer("Datenverarbeitung"):
    process_large_dataset()

Kontextmanager-Klassen

class DatabaseTransaction:
    def __init__(self, connection):
        self.connection = connection

    def __enter__(self):
        self.connection.begin_transaction()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.connection.commit()
        else:
            self.connection.rollback()
        return False  # Ausnahmen nicht unterdrücken

# Verwendung
with DatabaseTransaction(conn):
    user = conn.create_user(user_data)
    conn.create_profile(user.id, profile_data)

Comprehensions und Generatoren

Listen-Comprehensions

# Gut: Listen-Comprehension für einfache Transformationen
names = [user.name for user in users if user.is_active]

# Schlecht: Manuelle Schleife
names = []
for user in users:
    if user.is_active:
        names.append(user.name)

# Komplexe Comprehensions sollten erweitert werden
# Schlecht: zu komplex
result = [x * 2 for x in items if x > 0 if x % 2 == 0]

# Gut: Verwenden Sie eine Generator-Funktion
def filter_and_transform(items: Iterable[int]) -> list[int]:
    result = []
    for x in items:
        if x > 0 and x % 2 == 0:
            result.append(x * 2)
    return result

Generator-Ausdrücke

# Gut: Generator für faule Auswertung
total = sum(x * x for x in range(1_000_000))

# Schlecht: Erzeugt große Zwischenliste
total = sum([x * x for x in range(1_000_000)])

Generator-Funktionen

def read_large_file(path: str) -> Iterator[str]:
    """Liest eine große Datei zeilenweise."""
    with open(path) as f:
        for line in f:
            yield line.strip()

# Verwendung
for line in read_large_file("huge.txt"):
    process(line)

Datenklassen und Named Tuples

Datenklassen

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class User:
    """Benutzer-Entität mit automatischem __init__, __repr__ und __eq__."""
    id: str
    name: str
    email: str
    created_at: datetime = field(default_factory=datetime.now)
    is_active: bool = True

# Verwendung
user = User(
    id="123",
    name="Alice",
    email="[email protected]"
)

Datenklassen mit Validierung

@dataclass
class User:
    email: str
    age: int

    def __post_init__(self):
        # E-Mail-Format validieren
        if "@" not in self.email:
            raise ValueError(f"Ungültige E-Mail: {self.email}")
        # Altersbereich validieren
        if self.age < 0 or self.age > 150:
            raise ValueError(f"Ungültiges Alter: {self.age}")

Named Tuples

from typing import NamedTuple

class Point(NamedTuple):
    """Unveränderlicher 2D-Punkt."""
    x: float
    y: float

    def distance(self, other: 'Point') -> float:
        return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5

# Verwendung
p1 = Point(0, 0)
p2 = Point(3, 4)
print(p1.distance(p2))  # 5.0

Dekorateure

Funktionsdekorateure

import functools
import time

def timer(func: Callable) -> Callable:
    """Dekorateur zum Zeitmessen der Funktionsausführung."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} dauerte {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

# slow_function() gibt aus: slow_function dauerte 1.0012s

Parametrisierte Dekorateure

def repeat(times: int):
    """Dekorateur zum mehrmaligen Wiederholen einer Funktion."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(times):
                results.append(func(*args, **kwargs))
            return results
        return wrapper
    return decorator

@repeat(times=3)
def greet(name: str) -> str:
    return f"Hallo, {name}!"

# greet("Alice") gibt zurück: ["Hallo, Alice!", "Hallo, Alice!", "Hallo, Alice!"]

Klassenbasierte Dekorateure

class CountCalls:
    """Dekorateur, der zählt, wie oft eine Funktion aufgerufen wird."""
    def __init__(self, func: Callable):
        functools.update_wrapper(self, func)
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} wurde {self.count} Mal aufgerufen")
        return self.func(*args, **kwargs)

@CountCalls
def process():
    pass

# Jeder Aufruf von process() gibt die Aufrufanzahl aus

Nebenläufigkeitsmuster

Threading für I/O-gebundene Aufgaben

import concurrent.futures
import threading

def fetch_url(url: str) -> str:
    """Ruft eine URL ab (I/O-gebundener Vorgang)."""
    import urllib.request
    with urllib.request.urlopen(url) as response:
        return response.read().decode()

def fetch_all_urls(urls: list[str]) -> dict[str, str]:
    """Ruft mehrere URLs gleichzeitig mit Threads ab."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        results = {}
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                results[url] = future.result()
            except Exception as e:
                results[url] = f"Fehler: {e}"
    return results

Multiprocessing für CPU-gebundene Aufgaben

def process_data(data: list[int]) -> int:
    """CPU-intensive Berechnung."""
    return sum(x ** 2 for x in data)

def process_all(datasets: list[list[int]]) -> list[int]:
    """Verarbeitet mehrere Datensätze mit mehreren Prozessen."""
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = list(executor.map(process_data, datasets))
    return results

Async/Await für nebenläufiges I/O

import asyncio

async def fetch_async(url: str) -> str:
    """Ruft eine URL asynchron ab."""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def fetch_all(urls: list[str]) -> dict[str, str]:
    """Ruft mehrere URLs gleichzeitig ab."""
    tasks = [fetch_async(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return dict(zip(urls, results))

Paketorganisation

Standard-Projektstruktur

myproject/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── main.py
│       ├── api/
│       │   ├── __init__.py
│       │   └── routes.py
│       ├── models/
│       │   ├── __init__.py
│       │   └── user.py
│       └── utils/
│           ├── __init__.py
│           └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api.py
│   └── test_models.py
├── pyproject.toml
├── README.md
└── .gitignore

Import-Konventionen

# Gut: Import-Reihenfolge – Stdlib, Drittanbieter, lokal
import os
import sys
from pathlib import Path

import requests
from fastapi import FastAPI

from mypackage.models import User
from mypackage.utils import format_name

# Gut: Verwenden Sie isort für automatische Import-Sortierung
# pip install isort

init.py für Paket-Exporte

# mypackage/__init__.py
"""mypackage - Ein Beispiel-Python-Paket."""

__version__ = "1.0.0"

# Hauptklassen/-funktionen auf Paketebene exportieren
from mypackage.models import User, Post
from mypackage.utils import format_name

__all__ = ["User", "Post", "format_name"]

Speicher und Leistung

slots für Speichereffizienz

# Schlecht: Normale Klasse verwendet __dict__ (mehr Speicher)
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# Gut: __slots__ reduziert Speicherverbrauch
class Point:
    __slots__ = ['x', 'y']

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

Generator für große Datenmengen

# Schlecht: Gibt vollständige Liste im Speicher zurück
def read_lines(path: str) -> list[str]:
    with open(path) as f:
        return [line.strip() for line in f]

# Gut: Gibt Zeilen einzeln aus
def read_lines(path: str) -> Iterator[str]:
    with open(path) as f:
        for line in f:
            yield line.strip()

Zeichenkettenverkettung in Schleifen vermeiden

# Schlecht: O(n²) aufgrund der Unveränderlichkeit von Strings
result = ""
for item in items:
    result += str(item)

# Gut: O(n) mit join
result = "".join(str(item) for item in items)

# Gut: Verwenden Sie StringIO zum Aufbau
from io import StringIO

buffer = StringIO()
for item in items:
    buffer.write(str(item))
result = buffer.getvalue()

Integration von Python-Werkzeugen

Grundlegende Befehle

# Code-Formatierung
black .
isort .

# Linting
ruff check .

# Typüberprüfung
mypy src/

# Tests
pytest tests/

# Build
python -m build

# Abhängigkeiten verwalten
pip install -r requirements.txt
pip install -e .  # Entwicklungsmodus

pyproject.toml-Konfiguration

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.backends._legacy:_Backend"

[project]
name = "mypackage"
version = "1.0.0"
description = "Ein Beispiel-Python-Paket"
requires-python = ">=3.9"
dependencies = [
    "requests>=2.28.0",
    "fastapi>=0.100.0",
]

[tool.black]
line-length = 88
target-version = ["py39"]

[tool.isort]
profile = "black"
line_length = 88

[tool.ruff]
target-version = "py39"
line-length = 88
select = ["E", "F", "I", "W"]

[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]

Fehlerbehandlung (Fortsetzung)

Doppelt so viele Ressourcen oder Wiederherstellungsversuche

import time
from functools import wraps

def retry(max_attempts: int = 3, delay: float = 1.0):
    """Dekorateur zum Wiederholen einer Funktion bei Fehlschlag."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"Versuch {attempt + 1} fehlgeschlagen: {e}. Wiederholung in {delay}s...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0)
def unstable_network_call():
    # Simulierter instabiler Netzwerkaufruf
    pass

Tests und Qualitätssicherung

Tests mit pytest schreiben

# tests/test_user.py
import pytest
from mypackage.models import User

def test_user_creation():
    user = User(id="123", name="Alice", email="[email protected]")
    assert user.id == "123"
    assert user.name == "Alice"
    assert user.email == "[email protected]"
    assert user.is_active is True

def test_user_active_default():
    user = User(id="456", name="Bob", email="[email protected]")
    assert user.is_active is True

@pytest.mark.parametrize("email, expected", [
    ("[email protected]", True),
    ("invalid-email", False),
])
def test_email_validation(email, expected):
    # Angenommene Validierungslogik
    assert ("@" in email) == expected

class TestUserAPI:
    def test_get_user(self, client):
        response = client.get("/users/123")
        assert response.status_code == 200
        assert response.json()["id"] == "123"

Fixtures und Mocking

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from mypackage.main import app

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
def sample_user():
    return User(id="123", name="Alice", email="[email protected]")

# tests/test_service.py
from unittest.mock import patch

def test_external_api_call(mocker):
    mock_response = mocker.Mock()
    mock_response.json.return_value = {"status": "ok"}
    mocker.patch("requests.get", return_value=mock_response)
    
    result = call_external_api()
    assert result["status"] == "ok"

Nächste Schritte

1. Wenden Sie diese Muster auf bestehende Python-Projekte an
2. Richten Sie Linting und Formatierung mit ruff und black ein
3. Integrieren Sie mypy für statische Typüberprüfung in Ihrem Workflow
4. Führen Sie automatisierte Tests mit pytest und geeigneten Fixtures ein
5. Überprüfen und aktualisieren Sie regelmäßig Abhängigkeiten mit pip-audit oder safety

Zur weiteren Vertiefung:

• PEP 8 – Style Guide für Python-Code
• Python-Typannotationen
• Effizientes Python (Buch)
• Python-Paketierung