132 lines
3.8 KiB
Python
132 lines
3.8 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import Optional
|
|
from uuid import UUID, uuid4
|
|
|
|
from sqlmodel import Field, SQLModel, select, Relationship
|
|
from pydantic import Field as PydField
|
|
|
|
from account.models import Account
|
|
from account.schemas import AccountRead
|
|
from payee.models import Payee, PayeeRead
|
|
|
|
|
|
class TransactionBase(SQLModel):
|
|
transaction_date: date = Field()
|
|
payment_date: Optional[date] = Field(default=None)
|
|
|
|
class TransactionBaseId(TransactionBase):
|
|
id: UUID | None = Field(default_factory=uuid4, primary_key=True)
|
|
|
|
class Transaction(TransactionBaseId, table=True):
|
|
splits: list["Split"] = Relationship(back_populates="transaction")
|
|
|
|
@classmethod
|
|
def create(cls, transaction, session):
|
|
transaction_db = cls.model_validate(transaction)
|
|
session.add(transaction_db)
|
|
session.commit()
|
|
session.refresh(transaction_db)
|
|
|
|
return transaction_db
|
|
|
|
@classmethod
|
|
def list(cls):
|
|
return select(Transaction).join(Split).join(Account)
|
|
|
|
@classmethod
|
|
def get(cls, session, transaction_id):
|
|
return session.get(Transaction, transaction_id)
|
|
|
|
@classmethod
|
|
def update(cls, session, transaction_db, transaction_data):
|
|
transaction_db.sqlmodel_update(cls.model_validate(transaction_data))
|
|
session.add(transaction_db)
|
|
session.commit()
|
|
session.refresh(transaction_db)
|
|
return transaction_db
|
|
|
|
@classmethod
|
|
def delete(cls, session, transaction):
|
|
session.delete(transaction)
|
|
session.commit()
|
|
|
|
class SplitBase(SQLModel):
|
|
account_id: UUID = Field(foreign_key="account.id")
|
|
payee_id: Optional[UUID] = Field(foreign_key="payee.id")
|
|
amount: Decimal = Field(decimal_places=2)
|
|
|
|
class SplitBaseId(SplitBase):
|
|
transaction_id: UUID = Field(primary_key=True, foreign_key="transaction.id")
|
|
id: int = Field(primary_key=True)
|
|
|
|
class SplitRead(SplitBaseId):
|
|
account: AccountRead
|
|
payee: PayeeRead | None
|
|
|
|
class TransactionRead(TransactionBaseId):
|
|
splits: list[SplitRead]
|
|
|
|
class SplitWrite(SplitBase):
|
|
account_id: UUID = PydField(json_schema_extra={
|
|
"foreign_key": {
|
|
"reference": {
|
|
"resource": "accounts",
|
|
"schema": "AccountRead",
|
|
"label": "name"
|
|
}
|
|
}
|
|
})
|
|
payee_id: UUID | None = PydField(json_schema_extra={
|
|
"foreign_key": {
|
|
"reference": {
|
|
"resource": "payees",
|
|
"schema": "PayeeRead",
|
|
"label": "name"
|
|
}
|
|
}
|
|
})
|
|
amount: str = PydField(json_schema_extra={"format": "price-bgdc"})
|
|
|
|
class TransactionWrite(TransactionBase):
|
|
splits: list[SplitWrite] = PydField(json_schema_extra={"minItems": 1})
|
|
|
|
class TransactionCreate(TransactionWrite):
|
|
pass
|
|
|
|
class TransactionUpdate(TransactionWrite):
|
|
pass
|
|
|
|
class Split(SplitBaseId, table=True):
|
|
transaction: Transaction = Relationship(back_populates="splits")
|
|
account: Account | None = Relationship(back_populates="transaction_splits")
|
|
payee: Payee | None = Relationship()
|
|
|
|
@classmethod
|
|
def create(cls, transaction_split, session):
|
|
transaction_split_db = cls.model_validate(transaction_split)
|
|
session.add(transaction_split_db)
|
|
session.commit()
|
|
session.refresh(transaction_split_db)
|
|
|
|
return transaction_split_db
|
|
|
|
@classmethod
|
|
def update(cls, session, transaction_db, transaction_data):
|
|
transaction_db.sqlmodel_update(transaction_data)
|
|
session.add(transaction_db)
|
|
session.commit()
|
|
session.refresh(transaction_db)
|
|
return transaction_db
|
|
|
|
@classmethod
|
|
def delete(cls, session, transaction):
|
|
session.delete(transaction)
|
|
session.commit()
|
|
|
|
|
|
class SplitCreate(SplitWrite):
|
|
pass
|
|
|
|
class SplitUpdate(SplitWrite):
|
|
pass |