dodil/erp-platform

tests/scenario_bricks/scn_stock/infrastructure/tables.py

"""stock's one table, and the repository over it."""
from sqlalchemy import Column, Integer, MetaData, String, Table, TIMESTAMP, func, insert, select

md = MetaData()
reservations = Table("stock_reservations", md,
                     Column("reservation_key", String, primary_key=True), Column("sku", String),
                     Column("quantity", Integer), Column("reserved_by", String),
                     Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))


class StockRepository:
    def __init__(self, conn):
        self.conn = conn

    def get(self, key: str):
        r = self.conn.execute(select(reservations).where(reservations.c.reservation_key == key)).first()
        return dict(r._mapping) if r else None

    def add(self, row: dict) -> None:
        self.conn.execute(insert(reservations).values(**row))       # a plain INSERT: a second one of a key fails

    def totals(self, sku: str) -> tuple:
        r = self.conn.execute(select(func.count(), func.coalesce(func.sum(reservations.c.quantity), 0))
                              .where(reservations.c.sku == sku)).first()
        return int(r[0]), int(r[1])