from datetime import datetime, timezone from app.extensions import db from app.models.customer import Customer from app.models.account import Account from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import relationship from app.models.loan_charge import LoanCharge class Loan(db.Model): __tablename__ = 'loans' id = db.Column( db.Integer, primary_key=True, autoincrement=True, ) customer_id = db.Column(db.String(50), nullable=False) transaction_id = db.Column(db.String(50), nullable=True) account_id = db.Column(db.String(50), nullable=False) offer_id = db.Column(db.String(20), nullable=False) product_id = db.Column(db.String(20), nullable=True) collection_type = db.Column(db.String(20), nullable=True) current_loan_amount = db.Column(db.Float, nullable=True) initial_loan_amount = db.Column(db.Float, nullable=False) default_penalty_fee = db.Column(db.Float, default=0) continuous_fee = db.Column(db.Float, default=0) status = db.Column(db.String(20), default='pending') due_date = 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)) customer = relationship( "Customer", primaryjoin="Customer.id == Loan.customer_id", foreign_keys=[customer_id], back_populates="loans", ) loan_charges = relationship( "LoanCharge", primaryjoin="Loan.id == LoanCharge.loan_id", foreign_keys="LoanCharge.loan_id", back_populates="loan", ) @classmethod def create_loan(cls, customer_id, account_id, offer_id, product_id, initial_loan_amount, collection_type, transaction_id, status='pending'): # Check if customer exists customer = Customer.is_valid_customer(customer_id) if not customer: raise ValueError("Customer does not exist") now = datetime.now(timezone.utc) # Create and save the loan loan = cls( customer_id = customer_id, account_id = account_id, offer_id = offer_id, product_id = product_id, collection_type = collection_type, transaction_id = transaction_id, initial_loan_amount = initial_loan_amount, current_loan_amount = initial_loan_amount, due_date=now, status = status ) try: db.session.add(loan) except IntegrityError as err: raise ValueError(f"Database integrity error: {err}") return loan @classmethod def has_active_loans(cls, customer_id): active_loans = cls.query.filter_by( customer_id=customer_id, status='active' ).count() if active_loans > 0: return False return True @classmethod def get_customer_loan(cls, loan_id, customer_id): """ Get customer's active loans. """ loan = cls.query.filter_by(id = loan_id, customer_id = customer_id).first() if not loan: raise ValueError(f"Loan with ID {loan_id} does not exist or does not belong to customer {customer_id}.") return loan @classmethod def update_status(cls, loan_id, status): """ Update the status of the loan with the given loan_id. """ # Retrieve loan loan = cls.query.get(loan_id) if not loan: raise ValueError(f"Loan with ID {loan_id} does not exist.") if loan.status == status: return # Update loan status and the updated_at timestamp loan.status = status def to_dict(self): """ Convert the Loan object to a dictionary format for JSON serialization. """ return { 'debtId': self.id, 'initialLoanAmount': self.initial_loan_amount, 'currentLoanAmount': self.current_loan_amount, 'defaultPenaltyFee': self.default_penalty_fee, 'continuousFee': self.continuous_fee, 'collectionType': self.collection_type, 'status': self.status, 'dueDate': self.due_date.isoformat() if self.due_date else None, 'loanDate': self.created_at.isoformat() if self.created_at else None, } def __repr__(self): return f''