58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from datetime import datetime
|
|
from typing import Optional
|
|
from sqlalchemy.ext.hybrid import hybrid_property
|
|
|
|
from flaschengeist.database import db
|
|
from flaschengeist.models.user import User
|
|
from flaschengeist.models import ModelSerializeMixin, UtcDateTime, Serial
|
|
|
|
|
|
class Transaction(db.Model, ModelSerializeMixin):
|
|
__tablename__ = "balance_transaction"
|
|
# Protected foreign key properties
|
|
_receiver_id = db.Column("receiver_id", Serial, db.ForeignKey("user.id"))
|
|
_sender_id = db.Column("sender_id", Serial, db.ForeignKey("user.id"))
|
|
_author_id = db.Column("author_id", Serial, db.ForeignKey("user.id"), nullable=False)
|
|
|
|
# Public and exported member
|
|
id: int = db.Column("id", Serial, primary_key=True)
|
|
time: datetime = db.Column(UtcDateTime, nullable=False, default=UtcDateTime.current_utc)
|
|
amount: float = db.Column(db.Numeric(precision=5, scale=2, asdecimal=False), nullable=False)
|
|
reversal_id: Optional[int] = db.Column(Serial, db.ForeignKey("balance_transaction.id"))
|
|
|
|
# Dummy properties used for JSON serialization (userid instead of full user)
|
|
author_id: Optional[str] = None
|
|
sender_id: Optional[str] = None
|
|
original_id: Optional[int] = None
|
|
receiver_id: Optional[str] = None
|
|
|
|
# Not exported relationships just in backend only
|
|
sender_: User = db.relationship("User", foreign_keys=[_sender_id])
|
|
receiver_: User = db.relationship("User", foreign_keys=[_receiver_id])
|
|
author_: User = db.relationship("User", foreign_keys=[_author_id])
|
|
original_ = db.relationship("Transaction", uselist=False, backref=db.backref("reversal_", remote_side=[id]))
|
|
|
|
@hybrid_property
|
|
def sender_id(self):
|
|
return self.sender_.userid if self.sender_ else None
|
|
|
|
@sender_id.expression
|
|
def sender_id(cls):
|
|
return db.select([User.userid]).where(cls._sender_id == User.id_).scalar_subquery()
|
|
|
|
@hybrid_property
|
|
def receiver_id(self):
|
|
return self.receiver_.userid if self.receiver_ else None
|
|
|
|
@receiver_id.expression
|
|
def receiver_id(cls):
|
|
return db.select([User.userid]).where(cls._receiver_id == User.id_).scalar_subquery()
|
|
|
|
@property
|
|
def author_id(self):
|
|
return self.author_.userid
|
|
|
|
@property
|
|
def original_id(self):
|
|
return self.original_.id if self.original_ else None
|