first commit

app/models/__init__.py

@@ -0,0 +1,7 @@
+from .customer import Customer
+from .account import Account
+from .loan import Loan
+from .transaction import Transaction
+from .repayment import Repayment
+
+__all__ = ['Customer', 'Account', 'Loan', 'Transaction', 'Repayment']

app/models/account.py

@@ -0,0 +1,51 @@
+from datetime import datetime, timezone
+from sqlalchemy.orm import relationship
+from app.extensions import db
+from sqlalchemy.exc import IntegrityError
+
+class Account(db.Model):
+    __tablename__ = 'accounts'
+
+    id = db.Column(db.String(50), primary_key=True)  
+    customer_id = db.Column(db.String(50), nullable=False)
+    account_type = db.Column(db.String(50))
+    status = db.Column(db.String(20), default='active')
+    lien_amount = db.Column(db.Float, default=0.0)
+    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 == Account.customer_id",
+        foreign_keys=[customer_id],
+        back_populates="accounts",
+    )
+
+    @classmethod
+    def create_account(cls, id, customer_id, account_type, status='active'):
+        account = cls(
+            id=id,
+            customer_id=customer_id,
+            account_type=account_type
+        )
+
+        try:
+            db.session.add(account)
+            db.session.commit()
+        except IntegrityError as err:
+            db.session.rollback()
+            raise ValueError(f"Database integrity error: {err}")
+        return account
+
+    @classmethod
+    def is_valid_account(cls, account_id, customer_id):
+        account = cls.query.filter_by(id=account_id, customer_id=customer_id).first()
+        if not account:
+            return False
+        if account.lien_amount > 0:
+            return False
+        return True
+
+    def __repr__(self):
+        return f'<Account {self.id}>'
+    

app/models/customer.py

@@ -0,0 +1,54 @@
+from datetime import datetime, timezone
+from sqlalchemy.orm import relationship
+from app.extensions import db
+from app.models.account import Account
+from sqlalchemy.exc import IntegrityError
+
+class Customer(db.Model):
+    __tablename__ = 'customers'
+
+    id = db.Column(db.String(50), primary_key=True) 
+    msisdn = db.Column(db.String(20), unique=True, nullable=False)
+    country_code = db.Column(db.String(3), nullable=False)
+    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))
+
+    accounts = relationship(
+        "Account",
+        primaryjoin="Customer.id == Account.customer_id",
+        foreign_keys="Account.customer_id",
+        back_populates="customer",
+    )
+
+    @classmethod
+    def is_valid_customer(cls, customer_id):
+        customer = cls.query.filter_by(id=customer_id).first()
+        if not customer:
+            return False
+        return True
+
+    @classmethod
+    def create_customer(cls, id, msisdn, country_code, account_id, account_type='savings'):
+        if cls.query.filter_by(id=id).first():
+            raise ValueError("Customer already exists")
+        
+        # Create the customer
+        customer = cls(id=id, msisdn=msisdn, country_code=country_code)
+        try: 
+            db.session.add(customer)
+            
+            # Create an associated account
+            account = Account.create_account(
+                id=account_id,
+                customer_id=id,
+                account_type=account_type
+            )
+            
+            db.session.commit()
+        except IntegrityError as err:
+            db.session.rollback()
+            raise ValueError(f"Database integrity error: {err}")
+        return customer
+
+    def __repr__(self):
+        return f'<Customer {self.id}>'

app/models/loan.py

@@ -0,0 +1,99 @@
+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
+
+
+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)
+    account_id = db.Column(db.String(50), nullable=False)
+    offer_id = db.Column(db.String(20), nullable=False)
+    principal_amount = db.Column(db.Float, nullable=False)
+    status = db.Column(db.String(20), default='pending')
+    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))
+
+
+    @classmethod
+    def create_loan(cls, customer_id, account_id, offer_id, principal_amount, status='pending'):
+
+        # Check if customer exists
+        is_valid = Customer.is_valid_customer(customer_id)
+        if not is_valid:
+            raise ValueError("Customer does not exist")
+        
+        # # Check for active loans
+        # has_active_loans = cls.has_active_loans(customer_id)
+        # if has_active_loans:
+        #     raise ValueError("Customer has active loans")
+
+
+        # Create and save the loan
+        loan = cls(
+            customer_id=customer_id,
+            account_id=account_id,
+            offer_id=offer_id,
+            principal_amount=principal_amount,
+            status=status
+        )
+
+        try:
+            db.session.add(loan)
+            db.session.commit()
+        except IntegrityError as err:
+            db.session.rollback()
+            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
+        
+        db.session.commit()
+    
+    def __repr__(self):
+        return f'<Loan {self.id}>'

app/models/offer.py

@@ -0,0 +1,16 @@
+from datetime import datetime, timezone
+from app.extensions import db
+
+class Offer(db.Model):
+    __tablename__ = 'offers'
+
+    id = db.Column(db.Integer, primary_key=True)
+    product_id = db.Column(db.String, nullable=False)
+    min_amount = db.Column(db.Float, nullable=False)
+    max_amount = db.Column(db.Float, nullable=False)
+    tenor = db.Column(db.Integer, nullable=False)
+    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))
+
+    def __repr__(self):
+        return f'<LoanOffer {self.id}>'

app/models/repayment.py

@@ -0,0 +1,55 @@
+from datetime import datetime, timezone
+from app.api.enums.loan_status import LoanStatus
+from app.extensions import db
+from app.models.customer import Customer
+from app.models.loan import Loan
+from sqlalchemy.exc import IntegrityError
+
+
+class Repayment(db.Model):
+    __tablename__ = 'repayments'
+
+    id = db.Column(
+        db.Integer,
+        primary_key=True,
+        autoincrement=True,
+    )
+    loan_id = db.Column(db.String(50), nullable=False)
+    customer_id = db.Column(db.String(50), nullable=False)
+    product_id = db.Column(db.String(20), 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))
+
+    @classmethod
+    def create_repayment(cls, customer_id, loan_id, product_id):
+
+
+        # Check customer exists
+        if not Customer.is_valid_customer(customer_id):
+            raise ValueError("Invalid customer")
+
+        # Check loan exists
+        loan = Loan.get_customer_loan(loan_id = loan_id, customer_id = customer_id)
+        
+        # Check that the loan is active
+        if loan.status != LoanStatus.ACTIVE:
+            raise ValueError(f"Repayment cannot be processed. Loan status: ({loan.status})")
+        
+
+        repayment = cls(
+            customer_id=customer_id,
+            loan_id=loan.id,
+            product_id=product_id,
+        )
+
+        try:
+            db.session.add(repayment)
+            db.session.commit()
+        except IntegrityError as err:
+            db.session.rollback()
+            raise ValueError(f"Database integrity error: {err}")
+
+        return repayment
+
+    def __repr__(self):
+        return f'<Repayment {self.id}>'

app/models/transaction.py

@@ -0,0 +1,53 @@
+from datetime import datetime, timezone
+from app.extensions import db
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy import and_, or_, not_
+
+class Transaction(db.Model):
+    __tablename__ = 'transactions'
+    id = db.Column(
+        db.Integer,
+        primary_key=True,
+        autoincrement=True,
+    )
+    #id = db.Column(db.Int, primary_key=True)
+    transaction_id = db.Column(db.String(50), nullable=False)
+    account_id = db.Column(db.String(50), nullable=False)
+    type = db.Column(db.String(50), nullable=False)
+    channel = db.Column(db.String(50), nullable=False)
+    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))
+
+    def __repr__(self):
+        return f'<Transaction {self.id}>'
+
+    @classmethod
+    def create_transaction(cls, transaction_id, account_id, type, channel):
+
+        # if cls.query.filter_by(transaction_id=transaction_id).first():
+        #     raise ValueError("Duplicate Transaction")
+
+        if cls.query.filter( and_( cls.transaction_id ==transaction_id, cls.type==type) ).first():
+            raise ValueError("Duplicate Transaction")
+
+
+
+        transaction = cls(
+            transaction_id=transaction_id,
+            account_id=account_id,
+            type=type,
+            channel=channel
+        )
+
+        try:
+            db.session.add(transaction)
+            db.session.commit()
+        except IntegrityError as err:
+            db.session.rollback()
+            raise ValueError(f"Database integrity error: {err}")
+
+        return transaction
+
+    @classmethod
+    def get_transaction_by_id(cls, transaction_id):
+        return cls.query.get(transaction_id)