"""The tables, and the repository that loads and saves a Quote whole.
The mapping between rows and the domain is written by hand. That is the price of a domain with no
SQLAlchemy in it, and it is small: one function each way.
"""
from typing import Optional
from sqlalchemy import Column, Integer, MetaData, Numeric, String, Table, TIMESTAMP
from erp_platform.repository import Aggregate, Part
from brick_example.domain.quote import Line, Quote
md = MetaData()
quotes = Table("example_quotes", md,
Column("quote_id", String, primary_key=True), Column("customer", String),
Column("owner", String), Column("status", String), Column("currency", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
lines = Table("example_lines", md,
Column("quote_id", String, primary_key=True), Column("line_no", Integer, primary_key=True),
Column("description", String), Column("quantity", Integer),
Column("unit_price", Numeric(18, 2)),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
EXAMPLE = Aggregate(Part(quotes, ("quote_id",)), Part(lines, ("quote_id", "line_no")))
class QuoteRepository:
"""Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, quote_id: str) -> Optional[Quote]:
rows = EXAMPLE.load(self.conn, {"quote_id": quote_id})
if rows is None:
return None
head = rows["example_quotes"][0]
return Quote(quote_id=head["quote_id"], customer=head["customer"], owner=head["owner"],
currency=head["currency"], status=head["status"],
lines=sorted((Line(r["line_no"], r["description"], r["quantity"], r["unit_price"])
for r in rows["example_lines"]), key=lambda ln: ln.line_no))
def save(self, q: Quote) -> None:
EXAMPLE.save(self.conn, {
"example_quotes": [{"quote_id": q.quote_id, "customer": q.customer, "owner": q.owner,
"status": q.status, "currency": q.currency}],
"example_lines": [{"quote_id": q.quote_id, "line_no": ln.line_no,
"description": ln.description, "quantity": ln.quantity,
"unit_price": ln.unit_price} for ln in q.lines]})
def ids_owned_by(self, owner: Optional[str]) -> list:
from sqlalchemy import select
stmt = select(quotes.c.quote_id)
if owner is not None:
stmt = stmt.where(quotes.c.owner == owner)
return sorted(r[0] for r in self.conn.execute(stmt))