from datetime import datetime, timezone from app.extensions import db from sqlalchemy.orm import relationship # from dateutil.relativedelta import relativedelta class LoanRepaymentSchedule(db.Model): __tablename__ = 'loan_repayment_schedules' id = db.Column(db.Integer, primary_key=True, autoincrement=True) loan_id = db.Column(db.Integer, nullable=False) transaction_id = db.Column(db.String(50), nullable=True) product_id = db.Column(db.String(20), nullable=True) installment_number = db.Column(db.Integer, nullable=False) due_date = db.Column(db.DateTime, nullable=False) installment_amount= db.Column(db.Float, default=0.0) total_repayment_amount = db.Column(db.Float, default=0.0) paid = db.Column(db.Boolean, default=False) paid_at = db.Column(db.DateTime, nullable=True) created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc)) updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc)) # loan = relationship( # "Loan", # primaryjoin="LoanRepaymentSchedule.loan_id == Loan.id", # foreign_keys=[loan_id], # back_populates="loan_repayment_schedules", # ) @classmethod def get_all_repayment_schedules(cls, loan_id=None, product_id=None, transaction_id=None, paid=None, due_before=None, due_after=None, installment_number=None, page=1, limit=20): """ Get all loan repayment schedules with optional filtering Args: loan_id (int, optional): Filter by loan ID product_id (str, optional): Filter by product ID transaction_id (str, optional): Filter by transaction ID paid (bool, optional): Filter by paid status due_before (datetime, optional): Filter schedules due before this date due_after (datetime, optional): Filter schedules due after this date installment_number (int, optional): Filter by installment number page (int, optional): Page number for pagination limit (int, optional): Number of items per page Returns: tuple: (list of LoanRepaymentSchedule objects, total count) """ query = cls.query # Apply filters if provided if loan_id: query = query.filter(cls.loan_id == loan_id) if product_id: query = query.filter(cls.product_id == product_id) if transaction_id: query = query.filter(cls.transaction_id == transaction_id) if paid is not None: query = query.filter(cls.paid == paid) if due_before: query = query.filter(cls.due_date <= due_before) if due_after: query = query.filter(cls.due_date >= due_after) if installment_number: query = query.filter(cls.installment_number == installment_number) # Order by due_date and installment_number query = query.order_by(cls.due_date.asc(), cls.installment_number.asc()) # Get total count before pagination total_count = query.count() # Apply pagination offset = (page - 1) * limit query = query.limit(limit).offset(offset) return query.all(), total_count def to_dict(self): return { 'id': self.id, 'loan_id': self.loan_id, 'product_id': self.product_id, 'transaction_id': self.transaction_id, 'installment_number': self.installment_number, 'due_date': self.due_date.isoformat() if self.due_date else None, 'installment_amount': self.installment_amount, 'total_repayment_amount': self.total_repayment_amount, 'paid': self.paid, 'paid_at': self.paid_at.isoformat() if self.paid_at else None, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None } def __repr__(self): return f''