import json from typing import Dict from sqlalchemy import ( Column, Integer, ForeignKey, String, Table, Text, JSON, Float, DateTime, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.sql import func from sqlalchemy import event from src.apps.merchants.helper import get_brand_primary_color, get_brand_secondary_color from src.core.config import settings from src.core.utils import constants from src.apps.base.models.base import Base from src.core.utils.enums import ( PaymentCurrencies, SiteTemplateKeys, TransactionCategories, TransactionStatusTypes, TransactionTypes, TransactionSourceTypes, EmailTemplateIcons ) from src.worker.celery_app import celery_app transactions_invoices_map = Table( "transactions_invoices", Base.metadata, Column("transaction_id", Integer, ForeignKey("transactions.id"), primary_key=True), Column("invoice_id", Integer, ForeignKey("invoices.id"), primary_key=True), ) transactions_refunds_map = Table( "transactions_refunds", Base.metadata, Column("transaction_id", Integer, ForeignKey("transactions.id"), primary_key=True), Column("refund_id", Integer, ForeignKey("transactions.id"), primary_key=True), ) transactions_attachments_map = Table( "transactions_files", Base.metadata, Column("transaction_id", Integer, ForeignKey("transactions.id")), Column("file_id", Integer, ForeignKey("files.id")), ) transactions_notes_map = Table( "transactions_notes", Base.metadata, Column("transaction_id", Integer, ForeignKey("transactions.id")), Column("note_id", Integer, ForeignKey("notes.id")), ) class Transactions(Base): """ Transactions Model: ORM class for Transactions Entity """ __tablename__ = "transactions" id: Mapped[int] = mapped_column(primary_key=True, index=True, autoincrement=True) txn_amount: Mapped[float] = mapped_column(Float, default=0.0) txn_type: Mapped[str | None] = mapped_column(String(50), nullable=True) txn_status: Mapped[int] = mapped_column(default=TransactionStatusTypes.PAID) ocurred_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now()) txn_id: Mapped[str] = mapped_column(String(255), unique=True) txn_literal: Mapped[str | None] = mapped_column(Text, nullable=True) reference_id: Mapped[str | None] = mapped_column(Text, unique=True, nullable=True) currency: Mapped[str] = mapped_column(String(50), default=PaymentCurrencies.USD) platform_fee_amount: Mapped[float | None] = mapped_column(Float, nullable=True) txn_metadata: Mapped[dict | None] = mapped_column(JSON, nullable=True) billing_name: Mapped[str | None] = mapped_column(String(255), nullable=True) description: Mapped[str | None] = mapped_column(Text, nullable=True) category: Mapped[str] = mapped_column(String(50), default=TransactionCategories.CHARGE) transaction_type: Mapped[str] = mapped_column(String(50), default=TransactionTypes.PAYMENT_TERMINAL) txn_source: Mapped[str | None] = mapped_column(String(20), nullable=True) billing_address_id: Mapped[int | None] = mapped_column(ForeignKey("address.id"), nullable=True) billing_address: Mapped["Address"] = relationship("Address") payment_request_id: Mapped[int] = mapped_column(ForeignKey("payment_requests.id")) payment_request: Mapped["PaymentRequest"] = relationship("PaymentRequest") merchant_id: Mapped[int] = mapped_column(ForeignKey("merchants.id")) merchant: Mapped["Merchant"] = relationship("Merchant") customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id")) customer: Mapped["Customer"] = relationship("Customer") # invoice_id = Column(Integer, ForeignKey("invoices.id"), nullable=True) invoices: Mapped[list["Invoice"]] = relationship( "Invoice", secondary=transactions_invoices_map, back_populates="transactions" ) refunds: Mapped[list["Transactions"]] = relationship( "Transactions", secondary=transactions_refunds_map, primaryjoin=(transactions_refunds_map.c.transaction_id == id), secondaryjoin=(transactions_refunds_map.c.refund_id == id), ) payment_request_split_id: Mapped[int | None] = mapped_column(ForeignKey("split_payment_requests.id"), nullable=True) payment_request_split: Mapped["SplitPaymentRequests"] = relationship("SplitPaymentRequests") payment_method_id: Mapped[int | None] = mapped_column(ForeignKey("payment_methods.id"), nullable=True) payment_method: Mapped["PaymentMethod"] = relationship("PaymentMethod") attachments: Mapped[list["File"]] = relationship("File", secondary=transactions_attachments_map) notes: Mapped[list["Note"]] = relationship("Note", secondary=transactions_notes_map) @hybrid_property def txn_method(self) -> str: return "credit" @hybrid_property def status_text(self) -> str: if ( self.txn_status and self.txn_status >= TransactionStatusTypes.REFUNDED and self.txn_status < (TransactionStatusTypes.REFUNDED + 100) ): # return text 'refund' for all refund status types if self.category == TransactionCategories.REFUND: return "refund" return "paid" return next( ( " ".join(c.name.lower().split("_")) for c in TransactionStatusTypes if c.value == self.txn_status ), constants.DEFAULT_STATUS_TEXT, ) @hybrid_property def refundable_balance(self) -> float: balance: float = self.txn_amount if self.txn_status in [ TransactionStatusTypes.REFUNDED, TransactionStatusTypes.PENDING, TransactionStatusTypes.FAILED, ]: balance = 0 try: for refund in self.refunds: balance -= refund.txn_amount if balance < 0: balance = 0 break except: balance = 0 return balance @hybrid_property def additional_info(self) -> Dict: result: Dict = dict() try: if isinstance(self.txn_metadata, str): result = json.loads(self.txn_metadata) elif isinstance(self.txn_metadata, dict): result = self.txn_metadata except: pass return result # TRIGGERS @event.listens_for(Transactions.txn_status, "set") def receive_set(target, value, oldvalue, initiator): """listen for the 'set' event of txn_status column""" if target.category == TransactionCategories.REFUND and value != oldvalue: if value == TransactionStatusTypes.REFUNDED: celery_app.send_task( "src.apps.notification.tasks.task_send_email", kwargs={ "recepient": target.payment_method.payer.email, "template_key": SiteTemplateKeys.EMAIL_REFUND_SUCCEEDED, "template_data": { "email": target.payment_method.payer.email, "receiver_name": target.payment_method.payer.name, "merchant_id": ( target.merchant.merchant_id if target.merchant.merchant_id else None ), "currency": "$", "amount": "{:.2f}".format(target.txn_amount / 100), "template_primary_color": get_brand_primary_color( merchant_id=target.merchant.id ), "template_secondary_color": get_brand_secondary_color( merchant_id=target.merchant.id ), "template_logo": ( target.merchant.brand_logo.full_url if target.merchant.brand_logo else settings.APP_LOGO ), "transaction_id": target.txn_literal, "payment_request_id": target.payment_request.payment_request_literal, "merchant_name": target.merchant.name, "by_email": target.merchant.owner.email, "by_phone": target.merchant.owner.phone, "email_icon": f"{settings.SERVER_FRONT}{EmailTemplateIcons.REFUND_ICON}", "powered_by_icon": f"{settings.SERVER_FRONT}{EmailTemplateIcons.OWNER_LOGO}", }, }, )