Zum Inhalt
wannacode

Repository

Kurz gesagt: Repository versteckt die Datenbank hinter einem Interface mit Methoden wie find und save, sodass die Geschäftsregeln nichts von SQL wissen und der Speicher wechseln kann, ohne dass der Service es merkt.

Problem

Abfragen, verstreut über die Seite und die Kasse

Der Shop gibt einem Kunden mit drei Bestellungen oder mehr 10 % Rabatt. Die Seite des Kunden zeigt, ob es einen Rabatt gibt, und die Kasse wendet ihn beim Bezahlen an. Jede schreibt ihre eigene Abfrage an die Datenbank und ihre eigene Regel:

# Die Seite des Kunden
def show_discount(conn, customer):
    (count,) = conn.execute("SELECT COUNT(*) FROM orders WHERE customer = ?", (customer,)).fetchone()
    text = "10 % Rabatt" if count > 3 else "kein Rabatt"
    print(f"{customer}: {text}")




# Die Kasse
def checkout(conn, customer, total):
    rows = conn.execute("SELECT id FROM orders WHERE customer = ?", (customer,)).fetchall()
    return total * 0.9 if len(rows) >= 3 else total




conn = sqlite3.connect(":memory:")
setup(conn)
show_discount(conn, "Maria")
print(f"Maria zahlt {checkout(conn, 'Maria', 50.00):.2f} €")
conn.close()
Ausgabe
Maria: kein Rabatt
Maria zahlt 45.00 €

Maria hat drei Bestellungen. Ihre Seite sagt ihr „kein Rabatt“, weil dort die Regel count > 3 lautet. Die Kasse gibt ihr den Rabatt, weil dort count >= 3 steht. Zwei Stellen, zwei Versionen einer Regel, und niemand merkt es, bis sich ein Kunde beschwert.

Es gibt ein zweites Problem. Die Regel klebt an der Abfrage. Um zu prüfen, ob >= 3 stimmt, brauchst du eine Datenbank mit einer Tabelle und Zeilen darin. Die Tabelle orders und ihre Zeilen füllt die Funktion setup, wie in den Lektionen über die Datenbank im Programm. Hier ist sie versteckt, damit sie nicht stört.

Die Idee

Repository: die Daten hinter einem Interface

Die Abfragen wandern an eine Stelle: ein Repository. Es hat ein paar Methoden, die in der Sprache des Shops benannt sind, nicht in der der Datenbank: find_by_id, find_by_customer, save. Wer Bestellungen braucht, ruft sie auf und weiß nicht, ob dahinter SQL, eine Datei oder ein Dictionary im Speicher steckt.

Das ist das Repository-Muster. Drumherum ist der Code in Schichten geordnet, und jede Schicht spricht nur mit der Schicht unter ihr:

  • der Controller (die Seite, die Kasse) bekommt die Anfrage und zeigt die Antwort an;
  • der Service hält die Regel: „drei Bestellungen oder mehr geben 10 %“;
  • das Repository gibt und hält Daten. Regeln gibt es darin nicht;
  • die Daten: Speicher, SQLite oder eine andere Datenbank.

Der Service bekommt das Repository von außen, in seinem Konstruktor. Das ist Dependency Injection aus der Lektion Singleton und Dependency Injection. Darum kann jedes Repository mit denselben Methoden darunter sitzen.

In Python ist das Interface eher eine Vereinbarung als ein Typ: Der Service nimmt jedes Objekt, das find_by_customer hat. Die Basisklasse mit ABC schreibt die Vereinbarung nur auf und sagt dir früh, wenn eine Methode fehlt. Das Muster verschwindet nicht: Die Schichten und die Regel an einer Stelle bleiben dieselben.

Sieh es dir an

Der Vertrag und das Repository im Speicher

Die Bestellung und der Vertrag des Repositorys: drei Methoden und keine einzige Zeile SQL.

@dataclass(frozen=True)
class Order:
    id: int
    customer: str
    total: float




class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, order_id): ...


    @abstractmethod
    def find_by_customer(self, customer): ...


    @abstractmethod
    def save(self, order): ...

Das erste Repository hält die Bestellungen im Speicher, in einem Dictionary von der Nummer zur Bestellung. save mit derselben Nummer ersetzt die alte Bestellung.

class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self.orders = {}


    def find_by_id(self, order_id):
        return self.orders.get(order_id)


    def find_by_customer(self, customer):
        return [order for order in self.orders.values() if order.customer == customer]


    def save(self, order):
        self.orders[order.id] = order

Eine Funktion füllt jedes Repository mit den Bestellungen des Shops. Sie kennt nur den Vertrag, also funktioniert sie mit dem Speicher und mit der Datenbank:

def fill(orders):
    orders.save(Order(1, "Maria", 80.00))
    orders.save(Order(2, "Jonas", 25.50))
    orders.save(Order(3, "Maria", 40.00))
    orders.save(Order(4, "Maria", 12.90))

Sieh es dir an

Der Service weiß nicht, woher die Daten kommen

Die Regel steht im Service, und der Controller zeigt nur die Antwort an. Keiner von beiden schreibt SQL:

class LoyaltyService:
    def __init__(self, orders):
        self.orders = orders


    def discount_for(self, customer):
        return 10 if len(self.orders.find_by_customer(customer)) >= 3 else 0




class OrderController:
    def __init__(self, loyalty):
        self.loyalty = loyalty


    def show(self, customer):
        discount = self.loyalty.discount_for(customer)
        text = f"{discount} % Rabatt" if discount > 0 else "kein Rabatt"
        print(f"{customer}: {text}")

Derselbe Service und derselbe Controller, erst auf dem Speicher, dann auf SQLite:

memory = InMemoryOrderRepository()
fill(memory)
OrderController(LoyaltyService(memory)).show("Maria")


conn = sqlite3.connect(":memory:")
database = SqliteOrderRepository(conn)
fill(database)
OrderController(LoyaltyService(database)).show("Maria")
conn.close()
Ausgabe
Maria: 10 % Rabatt
Maria: 10 % Rabatt

class LoyaltyService:
    def __init__(self, orders):
        self.orders = orders


    def discount_for(self, customer):
        return 10 if len(self.orders.find_by_customer(customer)) >= 3 else 0




class OrderController:
    def __init__(self, loyalty):
        self.loyalty = loyalty


    def show(self, customer):
        discount = self.loyalty.discount_for(customer)
        text = f"{discount} % Rabatt" if discount > 0 else "kein Rabatt"
        print(f"{customer}: {text}")




memory = InMemoryOrderRepository()
fill(memory)
OrderController(LoyaltyService(memory)).show("Maria")


conn = sqlite3.connect(":memory:")
database = SqliteOrderRepository(conn)
fill(database)
OrderController(LoyaltyService(database)).show("Maria")
conn.close()
Schritt für Schritt, in Worten
  1. Ein Repository im Speicher, gefüllt mit den Bestellungen. Der Service bekommt es, ohne zu wissen, was es ist.
  2. Der Controller bekommt die Anfrage für Maria. Er fragt nicht die Datenbank: Er ruft den Service auf.
  3. Der Service braucht die Bestellungen des Kunden und fragt das Repository danach.
  4. find_by_customer auf OrderRepository. Das Repository im Speicher antwortet: Es geht sein Dictionary durch.
  5. Drei Bestellungen von Maria kommen nach oben. Die Regel steht hier und nur hier: 3 ≥ 3, also 10 %.
  6. Der Controller zeigt nur die Antwort an: „Maria: 10 % Rabatt“.
  7. Der Austausch: SQLite tritt an die Stelle des Speichers. Es ist dasselbe OrderRepository, darum werden der Service und der Controller nicht angefasst.
  8. Derselbe Aufruf. Nur das Objekt unter dem Service ist ein anderes.
  9. find_by_customer schickt jetzt SELECT … WHERE customer = ? an die Tabelle orders.
  10. Wieder drei Zeilen, wieder 10 %. Der Service hat nicht bemerkt, dass die Quelle eine andere ist.
  11. Dieselbe Zeile, diesmal aus einer Datenbank.
  12. Controller → Service → Repository → Daten. Nur die unterste Schicht hat sich geändert, und die Regel blieb an einer Stelle.

Nur die unterste Schicht hat sich geändert. Der Service und der Controller sind dieselben Klassen, ohne eine einzige geänderte Zeile, und Maria bekommt den Rabatt jetzt sowohl auf der Seite als auch an der Kasse, weil es eine einzige Regel gibt.

Sieh es dir an

Die echte Datenbank hinter demselben Interface

Hier ist das Repository in SQLite. Jede Methode ist eine Abfrage mit Parametern (?), nie Text, der in das SQL geklebt wird. Warum das wichtig ist, steht in der Lektion SQL-Injection und Parameter.

class SqliteOrderRepository(OrderRepository):
    def __init__(self, conn):
        self.conn = conn
        conn.execute("CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, customer TEXT NOT NULL, total REAL NOT NULL)")


    def find_by_id(self, order_id):
        row = self.conn.execute("SELECT id, customer, total FROM orders WHERE id = ?", (order_id,)).fetchone()
        return Order(*row) if row else None


    def find_by_customer(self, customer):
        rows = self.conn.execute("SELECT id, customer, total FROM orders WHERE customer = ? ORDER BY id", (customer,))
        return [Order(*row) for row in rows]


    def save(self, order):
        self.conn.execute(
            "INSERT INTO orders (id, customer, total) VALUES (?, ?, ?) ON CONFLICT (id) DO UPDATE SET customer = excluded.customer, total = excluded.total",
            (order.id, order.customer, order.total),
        )
Order(*row) macht aus der Zeile eine Bestellung: Die Spalten stehen in derselben Reihenfolge wie die Felder von Order.

Hier lebt das SQL für Bestellungen. Hier kannst du eine langsame Abfrage reparieren, einen Index hinzufügen oder SQLite gegen PostgreSQL tauschen, und der Service merkt es nicht. Wird das Repository zu einem ORM, achte auf eine Abfrage pro Zeile: Die Lektion ORM und das N+1-Problem zeigt, wie das aussieht.

Sieh es dir an

Die Regel ohne Datenbank

Weil der Service jedes Repository akzeptiert, prüfst du die Regel mit dem Repository im Speicher: keine Datei, kein Server, in Millisekunden.

orders = InMemoryOrderRepository()
loyalty = LoyaltyService(orders)
orders.save(Order(1, "Maria", 80.00))
orders.save(Order(2, "Maria", 40.00))
print(f"2 Bestellungen: {loyalty.discount_for('Maria')}%")
orders.save(Order(3, "Maria", 12.90))
print(f"3 Bestellungen: {loyalty.discount_for('Maria')}%")
Ausgabe
2 Bestellungen: 0%
3 Bestellungen: 10%

Zwei Bestellungen genügen nicht, die dritte gibt 10 %. Genau das ist die Grenze, an der die Seite und die Kasse sich nicht einig waren. So sieht auch ein Test der Regel aus: Du füllst das Repository im Speicher und prüfst, was der Service sagt.

Probier es aus

Was kommt heraus?

Die Bestellung mit der Nummer 1 wird zweimal gespeichert, mit einer anderen Summe. Was gibt das Programm aus?

orders = InMemoryOrderRepository()
orders.save(Order(1, "Maria", 80.00))
orders.save(Order(1, "Maria", 50.00))
print(len(orders.find_by_customer("Maria")))
print(f"{orders.find_by_id(1).total:.2f} €")
Antwort

Probier es aus

Finde den Fehler

Jemand findet die Suche nach Kunde langsam und schreibt ein Repository, das die Bestellungen gleich nach Kunde hält:

class FastOrderRepository(OrderRepository):
    def __init__(self):
        self.by_customer = {}


    def find_by_id(self, order_id):
        for orders in self.by_customer.values():
            for order in orders:
                if order.id == order_id:
                    return order
        return None


    def find_by_customer(self, customer):
        return self.by_customer.get(customer, [])


    def save(self, order):
        for orders in self.by_customer.values():
            orders[:] = [o for o in orders if o.id != order.id]
        self.by_customer.setdefault(order.customer, []).append(order)

Im Service taucht eine neue Methode auf: „Gibt es einen Rabatt, wenn er auch diese kauft?“ Dann fragt die Seite, wie viele Bestellungen Jonas hat:

class LoyaltyService:
    def __init__(self, orders):
        self.orders = orders


    def discount_for(self, customer):
        return 10 if len(self.orders.find_by_customer(customer)) >= 3 else 0


    def discount_if_buys(self, customer, cart):
        orders = self.orders.find_by_customer(customer)
        orders.append(cart)
        return 10 if len(orders) >= 3 else 0
orders = FastOrderRepository()
fill(orders)
loyalty = LoyaltyService(orders)
print(f"Wenn Jonas noch eine kauft: {loyalty.discount_if_buys('Jonas', Order(5, 'Jonas', 30.00))}%")
print(f"Bestellungen von Jonas: {len(orders.find_by_customer('Jonas'))}")
Ausgabe
Wenn Jonas noch eine kauft: 0%
Bestellungen von Jonas: 2

Jonas hat eine Bestellung, und das Repository sagt zwei. Niemand hat save aufgerufen. In welcher Zeile des Repositorys steckt der Fehler?

Wähle die Zeile mit dem Fehler.

Die Idee

Wann du es nicht brauchst

Repository fügt ein Interface und mindestens eine Klasse hinzu. Es lohnt sich nicht, wenn:

  • das Programm ein kleines Skript mit einer Abfrage ist. Die Schichten werden länger als die Arbeit selbst;
  • du ein ORM benutzt, das schon ein Repository liefert (das Modell in Django). Dein eigenes Interface, das nur Zeile für Zeile dorthin weiterleitet, ist eine weitere Datei ohne etwas Neues;
  • das Repository für jede Seite eine Methode bekommt: find_by_customer_and_status_order_by_total und zwanzig weitere davon. Dann sind die Abfragen wieder verstreut, nur mit längeren Namen.

Es lohnt sich, wenn die Regeln ohne Datenbank geprüft werden müssen, wenn die Quelle wechseln kann (eine Datenbank, eine Datei, ein anderer Dienst) oder wenn dieselben Abfragen an vielen Stellen benutzt werden.

Übung

Aus dem Repository löschen

Löse die Aufgabe in deinem Editor.

Löschen. Füge delete(order_id) zum Vertrag und zu beiden Repositorys hinzu. In SQLite ist das DELETE FROM orders WHERE id = ?. Zeige Marias Rabatt, lösche die Bestellung 4 und zeige ihn noch einmal, einmal mit dem Speicher und einmal mit der Datenbank.

Ausgabe
Maria: 10 % Rabatt
Maria: kein Rabatt
Maria: 10 % Rabatt
Maria: kein Rabatt

Weitere Aufgaben

  • Eine Regel nach Summe. Der Shop ändert die Regel: 10 % ab 150 € ausgegebenem Geld. Welche Klasse fasst du an? Prüfe es mit dem Repository im Speicher, ohne Datenbank.
  • Ein drittes Repository. Schreibe ein Repository, das die Bestellungen in einer Datei hält, eine pro Zeile. Der Service und der Controller müssen unverändert bleiben.
  • Eine Kopie bei der Suche. Repariere das Repository aus „Finde den Fehler“ und füge eine Prüfung hinzu: Nach discount_if_buys hat Jonas weiterhin 1 Bestellung.

Test

Teste dich

1. Wo lebt die Regel „3 Bestellungen oder mehr geben 10 % Rabatt“?
2. Warum lässt sich die Regel ohne Datenbank prüfen?
3. Das Repository gibt die Liste zurück, die es innen hält, und der Aufrufer fügt etwas hinzu. Was passiert?
4. Das Projekt benutzt ein ORM, das schon für jede Tabelle ein Repository liefert. Lohnt es sich, obendrauf ein eigenes Interface mit denselben Methoden zu setzen?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·