""" Transaction Services - Business logic for transaction operations """ import math from typing import List, Optional, Dict, Any from sqlalchemy import select, func, or_ from sqlalchemy.orm import Session, joinedload, selectinload from src.apps.transactions.models.transactions import Transactions from src.apps.merchants.models.merchant import Merchant from src.apps.customers.models.customer import Customer from src.apps.base.models.address import Address from src.apps.customers.models.customer_contact import CustomerContact from src.apps.files.models.file import File from src.apps.merchants.models.merchant_users import MerchantUsers from src.apps.users.models.user import User from src.apps.payment_methods.models.payment_methods import PaymentMethod from src.apps.payment_methods.models.payment_method_card_details import PaymentMethodCardDetails from src.apps.payment_methods.models.payment_method_ach_details import PaymentMethodAchDetails from src.apps.payment_methods.models.payment_method_cheque_details import PaymentMethodChequeDetails from src.core.config import settings from src.core.utils.enums import TransactionStatusTypes def _get_transactions_with_relations( db: Session, page: int = 1, per_page: int = 10, fields: Optional[List[str]] = None, search: Optional[str] = None, merchant_id: Optional[int] = None, ) -> tuple[List[Transactions], int]: """ Get transactions with optional relationships loaded. Args: db: Database session page: Page number (1-indexed) per_page: Items per page fields: List of relationship fields to load (e.g., ['customer', 'merchant', 'payment_method']) search: Search term for filtering merchant_id: Filter by merchant ID Returns: Tuple of (transactions list, total count) """ # Base query stmt = select(Transactions) # Apply relationship loading based on fields parameter if fields: if 'customer' in fields: # Load customer (many-to-one) with joinedload # Load customer's merchant and nested relationships stmt = stmt.options( joinedload(Transactions.customer) .joinedload(Customer.merchant) .selectinload(Merchant.addresses) ) stmt = stmt.options( joinedload(Transactions.customer) .joinedload(Customer.merchant) .joinedload(Merchant.brand_logo) ) stmt = stmt.options( joinedload(Transactions.customer) .joinedload(Customer.merchant) .selectinload(Merchant.users) .joinedload(MerchantUsers.user) ) # Load customer's collections with selectinload (more efficient for one-to-many) stmt = stmt.options( joinedload(Transactions.customer) .selectinload(Customer.addresses) ) stmt = stmt.options( joinedload(Transactions.customer) .selectinload(Customer.contacts) ) stmt = stmt.options( joinedload(Transactions.customer) .selectinload(Customer.attachments) ) if 'merchant' in fields: stmt = stmt.options(joinedload(Transactions.merchant)) if 'payment_method' in fields: # Load payment method with card/ach/cheque details stmt = stmt.options(joinedload(Transactions.payment_method)) if 'payment_request' in fields: stmt = stmt.options(joinedload(Transactions.payment_request)) if 'invoices' in fields: # Use selectinload for collections stmt = stmt.options(selectinload(Transactions.invoices)) if 'billing_address' in fields: stmt = stmt.options(joinedload(Transactions.billing_address)) # Apply filters if merchant_id: stmt = stmt.where(Transactions.merchant_id == merchant_id) if search: search_pattern = f"%{search}%" stmt = stmt.where( or_( Transactions.txn_id.ilike(search_pattern), Transactions.txn_literal.ilike(search_pattern), Transactions.reference_id.ilike(search_pattern), Transactions.billing_name.ilike(search_pattern), ) ) # Get total count count_stmt = select(func.count()).select_from(Transactions) if merchant_id: count_stmt = count_stmt.where(Transactions.merchant_id == merchant_id) if search: search_pattern = f"%{search}%" count_stmt = count_stmt.where( or_( Transactions.txn_id.ilike(search_pattern), Transactions.txn_literal.ilike(search_pattern), Transactions.reference_id.ilike(search_pattern), Transactions.billing_name.ilike(search_pattern), ) ) total = db.execute(count_stmt).scalar_one() # Apply pagination and ordering stmt = stmt.order_by(Transactions.ocurred_at.desc()) stmt = stmt.offset((page - 1) * per_page).limit(per_page) # Execute query - use unique() when eager loading collections transactions = db.execute(stmt).unique().scalars().all() return list(transactions), total def get_transaction_reports( db: Session, page: int = 1, per_page: int = 10, fields: Optional[List[str]] = None, search: Optional[str] = None, merchant_id: Optional[int] = None, ) -> Dict[str, Any]: """ Get paginated transaction reports with optional field selection. Args: db: Database session page: Page number (1-indexed) per_page: Items per page (max 100) fields: List of relationship fields to include (e.g., ['customer', 'merchant']) search: Search term merchant_id: Filter by merchant ID Returns: TransactionReportListResponseSchema with paginated results """ # Validate per_page limit per_page = min(per_page, 100) # Define allowed fields for inclusion allowed_fields = [ "merchant", "customer", "invoices", "payment_request", "billing_address", "payment_method" ] # Parse fields parameter requested_fields = [] if fields: requested_fields = [f.strip() for f in fields if f.strip() in allowed_fields] # Fetch transactions with relations transactions, total = _get_transactions_with_relations( db=db, page=page, per_page=per_page, fields=requested_fields, search=search, merchant_id=merchant_id ) # Convert to response format matching the API structure result = [] for txn in transactions: txn_dict = { "txn_amount": float(txn.txn_amount) if txn.txn_amount is not None else None, "txn_type": txn.txn_type, "txn_status": txn.txn_status, "ocurred_at": txn.ocurred_at.isoformat() if txn.ocurred_at else None, "txn_id": txn.txn_id, "txn_literal": txn.txn_literal, "reference_id": txn.reference_id, "description": txn.description, "currency": txn.currency, "platform_fee_amount": float(txn.platform_fee_amount) if txn.platform_fee_amount is not None else None, "billing_name": txn.billing_name, "merchant": None, "customer": None, "payment_request": None, "payment_method": None, "id": txn.id, "category": txn.category, "charge_type": None, "additional_info": txn.additional_info, "status_text": txn.status_text, "refundable_balance": float(txn.refundable_balance) if txn.refundable_balance is not None else None, "billing_address": None, "invoices": None, "auth_amount": None, "txn_source": txn.txn_source, } # Conditionally include relationships based on requested fields if 'customer' in requested_fields and txn.customer: customer = txn.customer txn_dict["customer"] = { "uin": customer.uin, "customer_type": customer.customer_type, "account_type": customer.account_type, "industry": customer.industry, "office_phone": customer.office_phone, "account_expires_on": customer.account_expires_on.isoformat() if customer.account_expires_on else None, "timezone": customer.timezone, "website": customer.website, "account_tax_id": customer.account_tax_id, "account_literal": customer.account_literal, "id": customer.id, "customer_id": customer.customer_id, "tilled_id": customer.tilled_id, "is_active": customer.is_active, "created_at": customer.created_at.isoformat() if customer.created_at else None, "updated_at": customer.updated_at.isoformat() if customer.updated_at else None, "deleted_at": customer.deleted_at.isoformat() if customer.deleted_at else None, "merchant": None, "avatar": None, "addresses": [], "tags_list": customer.tags.split(',') if customer.tags else [], "contacts": [], "default_address": {}, "billing_address": {}, "shipping_address": {}, "attachments": [], "first_name": customer.first_name, "email": customer.email, "phone": customer.phone, "business_legal_name": customer.business_legal_name, "last_active_date": txn.ocurred_at.isoformat() if txn.ocurred_at else None, } # Include addresses for addr in customer.addresses: addr_dict = { "id": addr.id, "address_line_1": addr.address_line_1, "address_line_2": addr.address_line_2 if addr.address_line_2 else "", "zipcode": addr.zipcode, "city": addr.city, "state": addr.state, "country": addr.country if addr.country else "US", "address_type": addr.address_type, } if hasattr(addr, 'use_as_default'): addr_dict["use_as_default"] = addr.use_as_default else: addr_dict["use_as_default"] = False txn_dict["customer"]["addresses"].append(addr_dict) # Include contacts for contact in customer.contacts: contact_name = None if hasattr(contact, 'first_name') and hasattr(contact, 'last_name'): contact_name = f"{contact.first_name or ''} {contact.last_name or ''}".strip() txn_dict["customer"]["contacts"].append({ "name": contact_name, "designation": contact.designation if hasattr(contact, 'designation') else None, "email": contact.email if hasattr(contact, 'email') else None, "phone": contact.phone if hasattr(contact, 'phone') else None, "id": contact.id, "contact_id": contact.contact_id if hasattr(contact, 'contact_id') else None, }) # Include attachments for attachment in customer.attachments: file_url = None if hasattr(attachment, 'path') and attachment.path: file_url = f"{settings.SERVER_HOST}/api/v1/files/download?filepath={attachment.path}" elif hasattr(attachment, 'file_path') and attachment.file_path: file_url = f"{settings.SERVER_HOST}/api/v1/files/download?filepath={attachment.file_path}" txn_dict["customer"]["attachments"].append({ "id": attachment.id, "original_name": attachment.original_name if hasattr(attachment, 'original_name') else None, "full_url": file_url, "created_at": attachment.created_at.isoformat() if hasattr(attachment, 'created_at') and attachment.created_at else None, }) # Include merchant within customer if loaded if customer.merchant: merchant = customer.merchant txn_dict["customer"]["merchant"] = { "name": merchant.name, "industry": merchant.industry, "tagline": merchant.tagline, "email": merchant.email, "phone": merchant.phone, "registration_no": merchant.registration_no, "license_no": merchant.license_no, "merchant_literal": merchant.merchant_literal, "id": merchant.id, "merchant_id": merchant.merchant_id, "uuid": str(merchant.uuid) if merchant.uuid else None, "tilled_account_id": merchant.tilled_account_id, "uin": merchant.uin, "is_active": merchant.is_active, "is_verified": merchant.is_verified, "is_onboarded": merchant.is_onboarded, "created_at": merchant.created_at.isoformat() if merchant.created_at else None, "updated_at": merchant.updated_at.isoformat() if merchant.updated_at else None, "deleted_at": merchant.deleted_at.isoformat() if merchant.deleted_at else None, "owner": None, "addresses": [], "business_profile": None, "logo_url": None, "active_domain": None, "default_address": {}, "subscriptions": [], } # Include merchant owner (from merchant_users relationship) if merchant.users: owner_user_map = next((mu for mu in merchant.users if mu.is_owner), None) if owner_user_map and owner_user_map.user: owner = owner_user_map.user txn_dict["customer"]["merchant"]["owner"] = { "first_name": owner.first_name, "middle_name": owner.middle_name if hasattr(owner, 'middle_name') else None, "last_name": owner.last_name, "email": owner.email, "phone": owner.phone if hasattr(owner, 'phone') else None, "id": owner.id, "user_id": owner.user_id if hasattr(owner, 'user_id') else None, "uuid": str(owner.uuid) if hasattr(owner, 'uuid') and owner.uuid else None, "is_active": owner.is_active, "is_verified": owner.is_verified if hasattr(owner, 'is_verified') else None, "access_level": getattr(owner, 'access_level', None), "last_login": owner.last_login.isoformat() if hasattr(owner, 'last_login') and owner.last_login else None, "dob": owner.dob.isoformat() if hasattr(owner, 'dob') and owner.dob else None, "timezone": getattr(owner, 'timezone', None), "created_at": owner.created_at.isoformat() if hasattr(owner, 'created_at') and owner.created_at else None, "updated_at": owner.updated_at.isoformat() if hasattr(owner, 'updated_at') and owner.updated_at else None, "deleted_at": owner.deleted_at.isoformat() if hasattr(owner, 'deleted_at') and owner.deleted_at else None, } # Include merchant addresses for addr in merchant.addresses: txn_dict["customer"]["merchant"]["addresses"].append({ "address_line_1": addr.address_line_1, "address_line_2": addr.address_line_2 if addr.address_line_2 else "", "zipcode": addr.zipcode, "city": addr.city, "state": addr.state, "country": addr.country if addr.country else "US", "address_type": addr.address_type, "id": addr.id, }) # Include merchant logo if merchant.brand_logo and hasattr(merchant.brand_logo, 'path'): txn_dict["customer"]["merchant"]["logo_url"] = f"{settings.SERVER_HOST}/api/v1/files/download?filepath={merchant.brand_logo.path}" elif merchant.brand_logo and hasattr(merchant.brand_logo, 'file_path'): txn_dict["customer"]["merchant"]["logo_url"] = f"{settings.SERVER_HOST}/api/v1/files/download?filepath={merchant.brand_logo.file_path}" # Include merchant active domain if merchant.subdomain: txn_dict["customer"]["merchant"]["active_domain"] = f"https://{merchant.subdomain}.fluxpay.com" elif merchant.custom_domain: txn_dict["customer"]["merchant"]["active_domain"] = f"https://{merchant.custom_domain}" # TODO: Include business_profile from business_profiles table # TODO: Include subscriptions from subscriptions table if 'payment_method' in requested_fields and txn.payment_method: pm = txn.payment_method pm_dict = { "method": pm.method, "id": pm.id, "is_connected": pm.is_connected, "payment_method_id": pm.payment_method_id, "scope": pm.scope, "card_details": None, "ach_details": None, "cheque_details": None, "created_at": pm.created_at.isoformat() if pm.created_at else None, "reference_id": pm.card_details.reference_id if pm.card_details and hasattr(pm.card_details, 'reference_id') else None, "is_default": pm.card_details.is_default if pm.card_details and hasattr(pm.card_details, 'is_default') else False, "is_deletable": False, "in_use": pm.in_use, "last_used_at": None, } # Include card details if available if pm.card_details: cd = pm.card_details pm_dict["card_details"] = { "reference_id": cd.reference_id, "is_default": cd.is_default, "card_number": cd.card_number, "brand": cd.brand, "cvv": cd.cvv, "funding": cd.funding, "expire_month": cd.expire_month, "expire_year": cd.expire_year, "last_used_at": cd.last_used_at.isoformat() if cd.last_used_at else None, } pm_dict["last_used_at"] = cd.last_used_at.isoformat() if cd.last_used_at else None # Include ACH details if available if pm.ach_details: ad = pm.ach_details pm_dict["ach_details"] = { "account_number": ad.account_number if hasattr(ad, 'account_number') else None, "routing_number": ad.routing_number if hasattr(ad, 'routing_number') else None, "account_type": ad.account_type if hasattr(ad, 'account_type') else None, "bank_name": ad.bank_name if hasattr(ad, 'bank_name') else None, } # Include cheque details if available if pm.cheque_details: chq = pm.cheque_details pm_dict["cheque_details"] = { "cheque_number": chq.cheque_number if hasattr(chq, 'cheque_number') else None, "bank_name": chq.bank_name if hasattr(chq, 'bank_name') else None, } txn_dict["payment_method"] = pm_dict result.append(txn_dict) # Calculate pagination metadata total_pages = math.ceil(total / per_page) if total > 0 else 1 base_url = f"{settings.SERVER_HOST}/api/v1/reports/transactions" # Build pagination URLs next_url = None prev_url = None first_url = f"{base_url}?page=1&per_page={per_page}" last_url = f"{base_url}?page={total_pages}&per_page={per_page}" if page < total_pages: next_url = f"{base_url}?page={page + 1}&per_page={per_page}" if page > 1: prev_url = f"{base_url}?page={page - 1}&per_page={per_page}" return { "allowed_fields": allowed_fields, "total": total, "page": page, "per_page": per_page, "next": next_url, "previous": prev_url, "first": first_url, "last": last_url, "result": result } def get_transaction_summary(db: Session) -> Dict[str, Any]: """ Get transaction summary statistics. Calculates: - Total transaction count and value - Declined transaction count and value (FAILED status) - Voided transaction count and value - Average transaction value Args: db: Database session Returns: Dict with transaction summary statistics """ # Query for all transactions summary total_stats = db.execute( select( func.count(Transactions.id).label('total_count'), func.coalesce(func.sum(Transactions.txn_amount), 0).label('total_value'), func.coalesce(func.avg(Transactions.txn_amount), 0).label('average_value') ) ).one() # Query for declined transactions (FAILED status = 300) declined_stats = db.execute( select( func.count(Transactions.id).label('declined_count'), func.coalesce(func.sum(Transactions.txn_amount), 0).label('declined_value') ) .where(Transactions.txn_status == TransactionStatusTypes.FAILED) ).one() # Query for voided transactions # Note: Checking if there's a voided status or if we need to use a different field voided_stats = db.execute( select( func.count(Transactions.id).label('voided_count'), func.coalesce(func.sum(Transactions.txn_amount), 0).label('voided_value') ) .where(Transactions.category == 'void') ).one() return { "total_transaction_count": total_stats.total_count, "total_transaction_value": float(total_stats.total_value), "total_declined_count": declined_stats.declined_count, "total_declined_value": float(declined_stats.declined_value), "average_transaction_value": float(total_stats.average_value), "voided_transaction_count": voided_stats.voided_count, "voided_transaction_value": float(voided_stats.voided_value) }