allmende-payment-system/src/allmende_payment_system/models.py

import datetime
import decimal
import typing

from sqlalchemy import Column, ForeignKey, Numeric, Table, select
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    object_session,
    relationship,
)

TABLE_PREFIX = "aps_"


class Base(DeclarativeBase):
    pass


class Account(Base):
    __tablename__ = TABLE_PREFIX + "account"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False, unique=True)

    users: Mapped[list["User"]] = relationship(
        "User",
        secondary=TABLE_PREFIX + "user_account_association",
        back_populates="accounts",
    )
    transactions: Mapped[list["Transaction"]] = relationship("Transaction")

    @property
    def balance(self):
        return sum(t.total_amount for t in self.transactions)


user_account_association = Table(
    TABLE_PREFIX + "user_account_association",
    Base.metadata,
    Column("user_id", ForeignKey(TABLE_PREFIX + "user.id"), primary_key=True),
    Column("account_id", ForeignKey(TABLE_PREFIX + "account.id"), primary_key=True),
)


class User(Base):
    __tablename__ = TABLE_PREFIX + "user"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(nullable=False, unique=True)
    display_name: Mapped[str] = mapped_column(nullable=False)

    accounts: Mapped[list["Account"]] = relationship(
        "Account", secondary=user_account_association, back_populates="users"
    )
    orders: Mapped[list["Order"]] = relationship("Order", back_populates="user")

    @property
    def shopping_cart(self):
        for order in self.orders:
            if order.transaction is None:
                cart = order
                break
        else:
            cart = Order(user=self)
            session = object_session(self)
            session.add(cart)

        return cart


class Area(Base):
    __tablename__ = TABLE_PREFIX + "area"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False, unique=True)
    description: Mapped[str] = mapped_column(nullable=True)
    image_path: Mapped[str] = mapped_column(nullable=True)

    products: Mapped[list["Product"]] = relationship("Product")


UnitsOfMeasure = typing.Literal[
    "g",
    "kg",
    "l",
    "piece",
]


class Product(Base):
    __tablename__ = TABLE_PREFIX + "product"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False, unique=True)

    price: Mapped[decimal.Decimal] = mapped_column(Numeric(10, 2))
    unit_of_measure: Mapped[UnitsOfMeasure] = mapped_column(nullable=False)
    # TODO: limit this to actually used vat rates?
    vat_rate: Mapped[decimal.Decimal] = mapped_column(Numeric(10, 2))

    area_id: Mapped[int] = mapped_column(ForeignKey(TABLE_PREFIX + "area.id"))
    area: Mapped["Area"] = relationship("Area", back_populates="products")

    image_path: Mapped[str] = mapped_column(nullable=True)


class Order(Base):
    __tablename__ = TABLE_PREFIX + "order"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey(TABLE_PREFIX + "user.id"))
    user: Mapped[User] = relationship("User", back_populates="orders")

    transaction: Mapped["Transaction | None"] = relationship("Transaction")

    items: Mapped[list["OrderItem"]] = relationship(
        "OrderItem", cascade="all, delete-orphan", back_populates="order"
    )

    @property
    def is_in_shopping_cart(self):
        return self.transaction is None

    @property
    def total_amount(self):
        return sum(item.total_amount for item in self.items)

    def finalize(self, account: Account):
        """
        Moves the order from the shopping cart to a given account
        and adds a transaction to the account.

        :param account: The account to which the order should be finalized
        :raises ValueError: If the order is already finalized or empty"""
        if not self.is_in_shopping_cart:
            raise ValueError("Order is already finalized.")

        if not self.items:
            raise ValueError("Cannot finalize an empty order.")

        assert account in self.user.accounts, "Account does not belong to user."

        # create a transaction for the order
        transaction = Transaction(
            type="order",
            total_amount=-self.total_amount,
            order=self,
            account=account,
        )
        session = object_session(self)
        session.add(transaction)


class OrderItem(Base):
    __tablename__ = TABLE_PREFIX + "order_item"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_id: Mapped[int] = mapped_column(ForeignKey(TABLE_PREFIX + "order.id"))
    order: Mapped[Order] = relationship("Order", back_populates="items")

    product_id: Mapped[int] = mapped_column(ForeignKey(TABLE_PREFIX + "product.id"))
    product: Mapped[Product] = relationship("Product")
    quantity: Mapped[decimal.Decimal] = mapped_column(nullable=False)
    total_amount: Mapped[decimal.Decimal] = mapped_column(
        Numeric(10, 2), nullable=False
    )


TransactionTypes = typing.Literal[
    "order",
    "deposit",
    "withdrawal",
    "expense",
]


class Transaction(Base):
    __tablename__ = TABLE_PREFIX + "transaction"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    type: Mapped[TransactionTypes] = mapped_column(nullable=False)
    quantity: Mapped[decimal.Decimal] = mapped_column(Numeric(10, 2), nullable=True)
    timestamp: Mapped[datetime.datetime] = mapped_column(
        nullable=False, default=datetime.datetime.now()
    )
    total_amount: Mapped[decimal.Decimal] = mapped_column(
        Numeric(10, 2), nullable=False
    )

    order_id: Mapped[int] = mapped_column(
        ForeignKey(TABLE_PREFIX + "order.id"), nullable=True
    )
    order: Mapped["Order"] = relationship("Order", back_populates="transaction")

    account_id: Mapped[int] = mapped_column(ForeignKey(TABLE_PREFIX + "account.id"))
    account: Mapped["Account"] = relationship("Account", back_populates="transactions")