248 lines
8.2 KiB
Python
248 lines
8.2 KiB
Python
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,
|
|
)
|
|
|
|
from allmende_payment_system.types import UnitsOfMeasure
|
|
|
|
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")
|
|
|
|
user_groups: Mapped[list["UserGroup"]] = relationship(
|
|
"UserGroup",
|
|
secondary=TABLE_PREFIX + "user_user_group_association",
|
|
back_populates="users",
|
|
)
|
|
|
|
@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
|
|
|
|
def has_permission(self, scope: str, action: str) -> bool:
|
|
for group in self.user_groups:
|
|
for permission in group.permissions:
|
|
if permission.scope == scope and permission.action == action:
|
|
return True
|
|
return False
|
|
|
|
|
|
class UserGroup(Base):
|
|
__tablename__ = TABLE_PREFIX + "user_group"
|
|
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)
|
|
|
|
permissions = relationship("Permission", back_populates="user_group")
|
|
|
|
users: Mapped[list["User"]] = relationship(
|
|
"User",
|
|
secondary=TABLE_PREFIX + "user_user_group_association",
|
|
back_populates="user_groups",
|
|
)
|
|
|
|
|
|
user_group_association = Table(
|
|
TABLE_PREFIX + "user_user_group_association",
|
|
Base.metadata,
|
|
Column("user_id", ForeignKey(TABLE_PREFIX + "user.id"), primary_key=True),
|
|
Column(
|
|
"user_group_id", ForeignKey(TABLE_PREFIX + "user_group.id"), primary_key=True
|
|
),
|
|
)
|
|
|
|
|
|
class Permission(Base):
|
|
__tablename__ = TABLE_PREFIX + "permission"
|
|
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
|
scope: Mapped[str] = mapped_column(nullable=False)
|
|
action: Mapped[str] = mapped_column(nullable=False)
|
|
|
|
user_group_id: Mapped[int] = mapped_column(
|
|
ForeignKey(TABLE_PREFIX + "user_group.id")
|
|
)
|
|
user_group: Mapped["UserGroup"] = relationship(
|
|
"UserGroup", back_populates="permissions"
|
|
)
|
|
|
|
|
|
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")
|
|
|
|
|
|
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)
|
|
allow_fractional: Mapped[bool] = mapped_column(nullable=False, default=True)
|
|
|
|
# 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
|
|
)
|
|
|
|
def update_quantity(self, new_quantity: decimal.Decimal):
|
|
if new_quantity <= 0:
|
|
raise ValueError("Quantity must be positive.")
|
|
|
|
self.quantity = new_quantity
|
|
self.total_amount = self.product.price * new_quantity
|
|
|
|
|
|
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")
|