MERMS/MermsCoreFlask
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Custom templates fix

Browse Source
This commit is contained in:
CHIEFSOFT\ameye
2025-09-20 20:00:24 -04:00
parent ff8513292b
commit 4d894d8214
12 changed files with 762 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SQL/site_data.sql
+29
View File
@@ -635,6 +635,10 @@ ALTER TABLE ONLY office_users
ALTER TABLE office_users OWNER TO merms_panel; ALTER TABLE office_users OWNER TO merms_panel;
INSERT INTO office_users (username, password, firstname, lastname, acc_level) VALUES ('mermsadmin','password','Merms','Admin',1000);
CREATE TABLE products_contacts( CREATE TABLE products_contacts(
id SERIAL, id SERIAL,
uid uuid DEFAULT uuid_generate_v4(), uid uuid DEFAULT uuid_generate_v4(),
@@ -654,6 +658,31 @@ ALTER TABLE ONLY products_contacts
ALTER TABLE products_contacts OWNER TO merms_panel; ALTER TABLE products_contacts OWNER TO merms_panel;
CREATE TABLE custom_templates (
id SERIAL,
uid uuid DEFAULT uuid_generate_v4(),
custom_id VARCHAR(25) UNIQUE NOT NULL,
provision_name VARCHAR(250),
added timestamp without time zone DEFAULT now()
);
ALTER TABLE ONLY custom_templates
ADD CONSTRAINT custom_templates_id_key UNIQUE (id);
ALTER TABLE custom_templates OWNER TO merms_panel;
INSERT INTO custom_templates(custom_id,provision_name) VALUES('icare_template','registry.chiefsoft.com/custom/mermsicare-mermshostedicare-001:latest');
INSERT INTO custom_templates(custom_id,provision_name) VALUES('kevkem_template','registry.chiefsoft.com/custom/mermskevkem-mermshostedicare-001:latest');
icare_template
registry.chiefsoft.com/custom/mermsicare-mermshostedicare-001:latest
kevkem_template
registry.chiefsoft.com/custom/mermskevkem-mermshostedicare-001:latest
new_l = { new_l = {
"uid": "425611f2-c692-4404-b93d-76ca7a5ce7" + str(x), "uid": "425611f2-c692-4404-b93d-76ca7a5ce7" + str(x),
"title": "Contact Random Item on " + str(x), "title": "Contact Random Item on " + str(x),
app/api/enums/__init__.py
+2 -1
View File
@@ -1,4 +1,5 @@
from .transaction_type import TransactionType from .transaction_type import TransactionType
from .loan_status import LoanStatus from .loan_status import LoanStatus
from .settings_items_data import SettingsItemsData from .settings_items_data import SettingsItemsData
from .generatives_list import GenerativesList from .generatives_list import GenerativesList
from .kafka_messages import KafkaMessage
app/api/enums/kafka_messages.py
+5
View File
@@ -0,0 +1,5 @@
from enum import Enum
class KafkaMessage(str, Enum):
REFRESH_PRODUCT_SETTINGS = "REFRESH_PRODUCT_SETTINGS"
FLUTTER_PAYMENT_RECEIVED = "FLUTTER_PAYMENT_RECEIVED"
app/api/services/base_service.py
+110 -144
View File
@@ -306,45 +306,45 @@ class BaseService:
logger.info(f"Processing {cls.TRANSACTION_TYPE} request") logger.info(f"Processing {cls.TRANSACTION_TYPE} request")
return schema.load(data) return schema.load(data)
@classmethod # @classmethod
def get_or_create_customer(cls, validated_data): # def get_or_create_customer(cls, validated_data):
#
""" # """
Check if a customer exists; if not, create one. # Check if a customer exists; if not, create one.
""" # """
#
# customer = Customer.query.filter_by(id=validated_data.get("customerId")).first()
# if not customer:
# customer = Customer.create_customer(
# id=validated_data.get("customerId"),
# msisdn=validated_data.get("msisdn"),
# country_code=validated_data.get("countryCode"),
# account_id=validated_data.get("accountId"),
# )
# return customer
customer = Customer.query.filter_by(id=validated_data.get("customerId")).first() # @classmethod
if not customer: # def validate_account_ownership(cls, account_id, customer_id):
customer = Customer.create_customer( # """
id=validated_data.get("customerId"), # Check if the provided account belongs to the customer.
msisdn=validated_data.get("msisdn"), # """
country_code=validated_data.get("countryCode"), # is_valid = Account.is_valid_account(account_id, customer_id)
account_id=validated_data.get("accountId"), # return is_valid
)
return customer
@classmethod # @classmethod
def validate_account_ownership(cls, account_id, customer_id): # def log_transaction(cls, validated_data):
""" # """
Check if the provided account belongs to the customer. # Create a new transaction.
""" # """
is_valid = Account.is_valid_account(account_id, customer_id) # channel = "USSD" if validated_data.get("channel") is None else validated_data.get("channel")
return is_valid #
# return Transaction.create_transaction(
@classmethod # transaction_id = validated_data.get("transactionId"),
def log_transaction(cls, validated_data): # customer_id = validated_data.get('customerId', None),
""" # account_id = validated_data.get("accountId", None),
Create a new transaction. # type = cls.TRANSACTION_TYPE,
""" # channel = channel,
channel = "USSD" if validated_data.get("channel") is None else validated_data.get("channel") # )
return Transaction.create_transaction(
transaction_id = validated_data.get("transactionId"),
customer_id = validated_data.get('customerId', None),
account_id = validated_data.get("accountId", None),
type = cls.TRANSACTION_TYPE,
channel = channel,
)
@classmethod @classmethod
def async_send_settings_refresh_to_kafka(cls, settings_data, subscription_uid, topic): def async_send_settings_refresh_to_kafka(cls, settings_data, subscription_uid, topic):
@@ -357,113 +357,79 @@ class BaseService:
KafkaIntegration.send_loan_request(loan_data = loan_data, request_id = request_id, topic = topic) KafkaIntegration.send_loan_request(loan_data = loan_data, request_id = request_id, topic = topic)
KafkaIntegration.flush() KafkaIntegration.flush()
#
@classmethod # @classmethod
def calculate_charges(cls, offer, amount): # def calculate_charges(cls, offer, amount):
""" # """
Calculates and returns the charges for the given offer and amount. # Calculates and returns the charges for the given offer and amount.
#
Args: # Args:
offer (Offer): The offer object that contains the charges. # offer (Offer): The offer object that contains the charges.
amount (float): The requested loan amount. # amount (float): The requested loan amount.
#
Returns: # Returns:
dict: A dictionary containing the calculated charges. # dict: A dictionary containing the calculated charges.
""" # """
if not offer or not offer.charges: # if not offer or not offer.charges:
logger.error(f"No charges found for offer ID {offer.id}") # logger.error(f"No charges found for offer ID {offer.id}")
return {"error": "No charges found for the offer"} # return {"error": "No charges found for the offer"}
#
loan_charges = offer.charges # loan_charges = offer.charges
tenor = offer.schedule # offer.tenor // 30 # Convert to months # tenor = offer.schedule # offer.tenor // 30 # Convert to months
interest = cls.get_charge_detail(rates = offer.interest_rate, charges = loan_charges, code = "INTEREST", amount = amount) # interest = cls.get_charge_detail(rates = offer.interest_rate, charges = loan_charges, code = "INTEREST", amount = amount)
management = cls.get_charge_detail(rates = offer.management_rate, charges = loan_charges, code = "MGTFEE", amount = amount) # management = cls.get_charge_detail(rates = offer.management_rate, charges = loan_charges, code = "MGTFEE", amount = amount)
insurance = cls.get_charge_detail(rates = offer.insurance_rate, charges = loan_charges, code = "INSURANCE", amount = amount) # insurance = cls.get_charge_detail(rates = offer.insurance_rate, charges = loan_charges, code = "INSURANCE", amount = amount)
vat = cls.get_charge_detail(rates = offer.vat_rate, charges = loan_charges, code = "VAT", amount = amount, management_fee = management["fee"]) # vat = cls.get_charge_detail(rates = offer.vat_rate, charges = loan_charges, code = "VAT", amount = amount, management_fee = management["fee"])
#
# Separate fees into upfront and postpaid # # Separate fees into upfront and postpaid
upfront_fees = [ # upfront_fees = [
fee["fee"] # fee["fee"]
for fee in [interest, management, insurance, vat] # for fee in [interest, management, insurance, vat]
if fee["due_days"] == 0 # if fee["due_days"] == 0
] # ]
#
postpaid_fees = [ # postpaid_fees = [
fee["fee"] # fee["fee"]
for fee in [interest, management, insurance, vat] # for fee in [interest, management, insurance, vat]
if fee["due_days"] != 0 # if fee["due_days"] != 0
] # ]
vat_test = vat["fee"] # vat_test = vat["fee"]
logger.info(f"VAT fee == *************** : {vat_test}") # logger.info(f"VAT fee == *************** : {vat_test}")
#
# Up-front payment: (only those fees due immediately i.e due_days == 0) # # Up-front payment: (only those fees due immediately i.e due_days == 0)
# upfront_payment = sum(upfront_fees) # # upfront_payment = sum(upfront_fees)
if offer.schedule == 1: # if offer.schedule == 1:
upfront_payment = vat["fee"] + management["fee"] + insurance["fee"] + interest["fee"] # upfront_payment = vat["fee"] + management["fee"] + insurance["fee"] + interest["fee"]
interest_amount = interest["fee"] # interest_amount = interest["fee"]
repayment_amount = amount # repayment_amount = amount
else: # else:
upfront_payment = vat["fee"] + insurance["fee"]+management["fee"] # upfront_payment = vat["fee"] + insurance["fee"]+management["fee"]
interest_amount = interest["fee"]*offer.schedule # interest_amount = interest["fee"]*offer.schedule
repayment_amount = amount + interest_amount # repayment_amount = amount + interest_amount
#
#
# # Repayment amount: (principal + only those fees not due immediately i.e due_days != 0)
# # repayment_amount = amount + (sum(postpaid_fees) * tenor)
#
# # Total amount: (upfront_payment + repayment_amount)
# total_amount = upfront_payment + repayment_amount
#
# # Calculate the installment amount
# installment_amount = repayment_amount / offer.schedule
#
# return {
# "interest": interest,
# "interest_amount": interest_amount,
# "management": management,
# "insurance": insurance,
# "vat": vat,
# "upfront_payment": round(upfront_payment, 2),
# "repayment_amount": round(repayment_amount, 2),
# "installment_amount": round(installment_amount, 2),
# "total_amount": round(total_amount, 2)
# }
#
#
# Repayment amount: (principal + only those fees not due immediately i.e due_days != 0)
# repayment_amount = amount + (sum(postpaid_fees) * tenor)
# Total amount: (upfront_payment + repayment_amount)
total_amount = upfront_payment + repayment_amount
# Calculate the installment amount
installment_amount = repayment_amount / offer.schedule
return {
"interest": interest,
"interest_amount": interest_amount,
"management": management,
"insurance": insurance,
"vat": vat,
"upfront_payment": round(upfront_payment, 2),
"repayment_amount": round(repayment_amount, 2),
"installment_amount": round(installment_amount, 2),
"total_amount": round(total_amount, 2)
}
@classmethod
def get_charge_detail(cls, rates, charges, code, amount, management_fee= 0.0):
"""
Get details for a specific charge code from a list of loan charges.
Returns default values if not found.
"""
fee = 0.0
if code == "VAT" and management_fee > 0:
fee = management_fee * rates / 100
else:
fee = amount * rates / 100
return {
"rate": rates,
"fee": round(fee, 2),
"due_days": 30,
"code": code,
"description" : "have no idea how to get this yet"
}
# if charge.code == code:
# if code == "VAT" and management_fee > 0:
# fee = management_fee * rates / 100
# else:
# fee = amount * rates / 100
#
# return {
# "rate": rates,
# "fee": round(fee, 2),
# "due_days": charge.due
# }
# return {"rate": 0, "fee": 0, "due_days": 0}
app/api/services/customer_consent.py
+59 -59
View File
@@ -1,60 +1,60 @@
from flask import request, jsonify # from flask import request, jsonify
from app.api.helpers.response_helper import ResponseHelper # from app.api.helpers.response_helper import ResponseHelper
# # from app.api.services.base_service import BaseService
# from marshmallow import ValidationError
# from app.utils.logger import logger
# # from app.api.schemas.customer_consent import CustomerConsentSchema
# from app.api.services.base_service import BaseService # from app.api.services.base_service import BaseService
from marshmallow import ValidationError # from app.api.enums import TransactionType
from app.utils.logger import logger # from app.extensions import db
# from app.api.schemas.customer_consent import CustomerConsentSchema #
from app.api.services.base_service import BaseService #
from app.api.enums import TransactionType # class CustomerConsentService(BaseService):
from app.extensions import db # TRANSACTION_TYPE = TransactionType.CUSTOMER_CONSENT
#
# @staticmethod
class CustomerConsentService(BaseService): # def process_request(data):
TRANSACTION_TYPE = TransactionType.CUSTOMER_CONSENT # """
# Process the CustomerConsent request.
@staticmethod #
def process_request(data): # Args:
""" # data (dict): The request data.
Process the CustomerConsent request. #
# Returns:
Args: # dict: A standardized response.
data (dict): The request data. # """
# try:
Returns: # with db.session.begin():
dict: A standardized response. # validated_data = CustomerConsentService.validate_data(data, CustomerConsentSchema())
""" # account_id = validated_data.get('accountId')
try: # customer_id = validated_data.get('customerId')
with db.session.begin(): #
validated_data = CustomerConsentService.validate_data(data, CustomerConsentSchema()) # if(CustomerConsentService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
account_id = validated_data.get('accountId') #
customer_id = validated_data.get('customerId') # transaction = CustomerConsentService.log_transaction(validated_data = validated_data)
#
if(CustomerConsentService.validate_account_ownership(account_id = account_id, customer_id = customer_id)): # if not transaction:
# logger.error(f"Failed to log transaction")
transaction = CustomerConsentService.log_transaction(validated_data = validated_data) # return ResponseHelper.error(result_description="Failed to log transaction.")
# else:
if not transaction: # return ResponseHelper.error(result_description="Invalid Customer or Account")
logger.error(f"Failed to log transaction") #
return ResponseHelper.error(result_description="Failed to log transaction.") #
else: # db.session.commit()
return ResponseHelper.error(result_description="Invalid Customer or Account") # return ResponseHelper.success(result_description="Request is received")
#
# except ValidationError as err:
db.session.commit() #
return ResponseHelper.success(result_description="Request is received") # logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
# db.session.rollback()
except ValidationError as err: # return ResponseHelper.unprocessable_entity(result_description="Validation exception")
#
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}") # except ValueError as err:
db.session.rollback() # logger.error(f"{getattr(err, 'messages', str(err))}")
return ResponseHelper.unprocessable_entity(result_description="Validation exception") # db.session.rollback()
# return ResponseHelper.error(result_description=str(err))
except ValueError as err: #
logger.error(f"{getattr(err, 'messages', str(err))}") # except Exception as e:
db.session.rollback() # logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(result_description=str(err)) # db.session.rollback()
# return ResponseHelper.internal_server_error()
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
db.session.rollback()
return ResponseHelper.internal_server_error()
app/api/services/myproduct.py
+2 -10
View File
@@ -2,7 +2,7 @@ from urllib import request
from flask import session, jsonify from flask import session, jsonify
from app.api.enums import SettingsItemsData from app.api.enums import SettingsItemsData,KafkaMessage
from app.api.schemas.myproduct_set_template import MyProductSetTemplateSchema from app.api.schemas.myproduct_set_template import MyProductSetTemplateSchema
from app.utils.logger import logger from app.utils.logger import logger
from app.api.services.base_service import BaseService from app.api.services.base_service import BaseService
@@ -279,15 +279,7 @@ class MyProductsService(BaseService):
# logger.info(f"Incoming MyProduct data ==>>>> {memberSubscription}") # logger.info(f"Incoming MyProduct data ==>>>> {memberSubscription}")
productDataStatus = memberSubscription.status productDataStatus = memberSubscription.status
product_subscription_uid = memberSubscription.uid product_subscription_uid = memberSubscription.uid
# product_subscription_external_url = memberSubscription.external_url
# product_subscription_internal_url = memberSubscription.internal_url
# result_data = {
# "myproudct": {
# "result": "Reveived under development ",
# "message": "to be fixed"
# }
# }
for key in settings.keys(): for key in settings.keys():
setting_value = settings[key] setting_value = settings[key]
@@ -303,7 +295,7 @@ class MyProductsService(BaseService):
} }
logger.error(f"Going for Thread ******************** ") logger.error(f"Going for Thread ******************** ")
thread = Thread(target=MyProductsService.async_send_settings_refresh_to_kafka, thread = Thread(target=MyProductsService.async_send_settings_refresh_to_kafka,
args=(response_data, subscription_uid, "REFRESH_PRODUCT_SETTINGS")) args=(response_data, subscription_uid, KafkaMessage.REFRESH_PRODUCT_SETTINGS))
thread.start() thread.start()
logger.error(f"After the Thread ******************** ") logger.error(f"After the Thread ******************** ")
app/models/__init__.py
+2
View File
@@ -18,10 +18,12 @@ from .payments import Payments
from .subscription_generative import SubscriptionGenerative from .subscription_generative import SubscriptionGenerative
from .generative_results import GenerativeResults from .generative_results import GenerativeResults
from .office_users import OfficeUsers from .office_users import OfficeUsers
from .custom_templates import CustomTemplates
__all__ = ['Members', 'Account', 'Products', __all__ = ['Members', 'Account', 'Products',
'MembersProducts', 'MembersActions', 'MembersPending', 'ProductsDetails', 'MembersProducts', 'MembersActions', 'MembersPending', 'ProductsDetails',
'ProvisionActions', 'MembersProductsRefresh', 'MembersProductsSettings', 'ProvisionActions', 'MembersProductsRefresh', 'MembersProductsSettings',
'PasswordReset', 'MembersProfile', 'SubscriptionOptions', 'SubscriptionOptionsItems', 'PasswordReset', 'MembersProfile', 'SubscriptionOptions', 'SubscriptionOptionsItems',
'ProductsTemplates', 'Payments', 'PaymentsSession', 'SubscriptionGenerative', 'GenerativeResults', 'ProductsTemplates', 'Payments', 'PaymentsSession', 'SubscriptionGenerative', 'GenerativeResults',
'CustomTemplates',
'OfficeUsers'] 'OfficeUsers']
app/models/custom_templates.py
+35
View File
@@ -0,0 +1,35 @@
from datetime import datetime, timezone
from app.extensions import db
from app.models.charge import Charge
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
class CustomTemplates(db.Model):
__tablename__ = 'custom_templates'
id = db.Column(db.String, primary_key=True)
uid = db.Column(db.String, nullable=False)
custom_id = db.Column(db.String, nullable=False)
provision_name = db.Column(db.String, nullable=False)
added = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
@classmethod
def get_template_by_custom_id(cls, custom_id):
templates = cls.query.filter_by(custom_id=str(custom_id)).all()
if not templates:
raise ValueError(f"Templates with Custom ID {custom_id} not found")
return templates
def to_dict(self):
return {
"id": self.id,
"uid": self.uid,
"custom_id": self.product_id,
"provision_name": self.name,
"added": self.added.isoformat() if self.added else None
}
def __repr__(self):
return f'<CustomTemplates {self.id}>'
app/models/loan.py
+261 -261
View File
@@ -1,261 +1,261 @@
from datetime import datetime, timezone, timedelta # from datetime import datetime, timezone, timedelta
from itertools import product # from itertools import product
from app.extensions import db # from app.extensions import db
from app.models.customer import Customer # from app.models.customer import Customer
from app.models.account import Account # from app.models.account import Account
from sqlalchemy.exc import IntegrityError # from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import relationship # from sqlalchemy.orm import relationship
from dateutil.relativedelta import relativedelta # from dateutil.relativedelta import relativedelta
from datetime import timedelta # from datetime import timedelta
import logging # import logging
from sqlalchemy import and_, or_, not_ # from sqlalchemy import and_, or_, not_
from sqlalchemy.sql import func # from sqlalchemy.sql import func
#
logger = logging.getLogger(__name__) # logger = logging.getLogger(__name__)
#
#
class Loan(db.Model): # class Loan(db.Model):
__tablename__ = 'loans' # __tablename__ = 'loans'
#
id = db.Column( # id = db.Column(
db.Integer, # db.Integer,
primary_key=True, # primary_key=True,
autoincrement=True, # autoincrement=True,
) # )
customer_id = db.Column(db.String(50), nullable=False) # customer_id = db.Column(db.String(50), nullable=False)
transaction_id = db.Column(db.String(50), nullable=True) # transaction_id = db.Column(db.String(50), nullable=True)
original_transaction = db.Column(db.String(50), nullable=True) # original_transaction = db.Column(db.String(50), nullable=True)
account_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) # offer_id = db.Column(db.String(20), nullable=False)
product_id = db.Column(db.String(20), nullable=True) # product_id = db.Column(db.String(20), nullable=True)
collection_type = 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) # current_loan_amount = db.Column(db.Float, nullable=True)
initial_loan_amount = db.Column(db.Float, nullable=False) # initial_loan_amount = db.Column(db.Float, nullable=False)
default_penalty_fee = db.Column(db.Float, default=0) # default_penalty_fee = db.Column(db.Float, default=0)
continuous_fee = db.Column(db.Float, default=0) # continuous_fee = db.Column(db.Float, default=0)
upfront_fee = db.Column(db.Float, nullable=True, default=0.0) # upfront_fee = db.Column(db.Float, nullable=True, default=0.0)
repayment_amount = db.Column(db.Float, nullable=True, default=0.0) # repayment_amount = db.Column(db.Float, nullable=True, default=0.0)
installment_amount = db.Column(db.Float, nullable=True, default=0.0) # installment_amount = db.Column(db.Float, nullable=True, default=0.0)
status = db.Column(db.String(20), default='pending') # status = db.Column(db.String(20), default='pending')
tenor = db.Column(db.Integer, nullable=True) # tenor = db.Column(db.Integer, nullable=True)
due_date = db.Column(db.DateTime, nullable=True) # due_date = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime(timezone=True), server_default=func.now()) # created_at = db.Column(db.DateTime(timezone=True), server_default=func.now())
updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) # updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
eligible_amount = db.Column(db.Float, nullable=True, default=0.0) # eligible_amount = db.Column(db.Float, nullable=True, default=0.0)
disburse_date = db.Column(db.DateTime, nullable=True) # disburse_date = db.Column(db.DateTime, nullable=True)
disburse_verify = db.Column(db.DateTime, nullable=True) # disburse_verify = db.Column(db.DateTime, nullable=True)
reference = db.Column(db.String(50), nullable=True) # reference = db.Column(db.String(50), nullable=True)
disburse_result = db.Column(db.String(10), nullable=True) # disburse_result = db.Column(db.String(10), nullable=True)
disburse_description = db.Column(db.String(100), nullable=True) # disburse_description = db.Column(db.String(100), nullable=True)
verify_result = db.Column(db.String(10), nullable=True) # verify_result = db.Column(db.String(10), nullable=True)
verify_description = db.Column(db.String(100), nullable=True) # verify_description = db.Column(db.String(100), nullable=True)
#
# customer = relationship( # # customer = relationship(
# "Customer", # # "Customer",
# primaryjoin="Customer.id == Loan.customer_id", # # primaryjoin="Customer.id == Loan.customer_id",
# foreign_keys=[customer_id], # # foreign_keys=[customer_id],
# back_populates="loans", # # back_populates="loans",
# ) # # )
#
loan_charges = relationship( # loan_charges = relationship(
"LoanCharge", # "LoanCharge",
primaryjoin="LoanCharge.loan_id == Loan.id", # primaryjoin="LoanCharge.loan_id == Loan.id",
foreign_keys="LoanCharge.loan_id", # foreign_keys="LoanCharge.loan_id",
back_populates="loan", # back_populates="loan",
) # )
#
loan_repayment_schedules = relationship( # loan_repayment_schedules = relationship(
"LoanRepaymentSchedule", # "LoanRepaymentSchedule",
primaryjoin="LoanRepaymentSchedule.loan_id == Loan.id", # primaryjoin="LoanRepaymentSchedule.loan_id == Loan.id",
foreign_keys="LoanRepaymentSchedule.loan_id", # foreign_keys="LoanRepaymentSchedule.loan_id",
back_populates="loan", # back_populates="loan",
) # )
#
#
@classmethod # @classmethod
def create_loan( # def create_loan(
cls, # cls,
customer_id, # customer_id,
account_id, # account_id,
offer_id, # offer_id,
product_id, # product_id,
initial_loan_amount, # initial_loan_amount,
collection_type, # collection_type,
transaction_id, # transaction_id,
original_transaction, # original_transaction,
upfront_fee, # upfront_fee,
repayment_amount, # repayment_amount,
installment_amount, # installment_amount,
tenor, # tenor,
eligible_amount, # eligible_amount,
reference, # reference,
status = "pending", # status = "pending",
): # ):
# Check if customer exists # # Check if customer exists
customer = Customer.is_valid_customer(customer_id) # customer = Customer.is_valid_customer(customer_id)
if not customer: # if not customer:
raise ValueError("Customer does not exist") # raise ValueError("Customer does not exist")
#
now = datetime.now(timezone.utc) # now = datetime.now(timezone.utc)
due_date = now + timedelta(days=tenor) # due_date = now + timedelta(days=tenor)
#
# Create and save the loan # # Create and save the loan
loan = cls( # loan = cls(
customer_id = customer_id, # customer_id = customer_id,
account_id = account_id, # account_id = account_id,
offer_id = offer_id, # offer_id = offer_id,
product_id = product_id, # product_id = product_id,
collection_type = collection_type, # collection_type = collection_type,
transaction_id = transaction_id, # transaction_id = transaction_id,
original_transaction = original_transaction, # original_transaction = original_transaction,
initial_loan_amount = initial_loan_amount, # initial_loan_amount = initial_loan_amount,
current_loan_amount = initial_loan_amount, # current_loan_amount = initial_loan_amount,
upfront_fee = upfront_fee, # upfront_fee = upfront_fee,
repayment_amount = repayment_amount, # repayment_amount = repayment_amount,
installment_amount = installment_amount, # installment_amount = installment_amount,
due_date=due_date, # due_date=due_date,
tenor = tenor, # tenor = tenor,
status = status, # status = status,
eligible_amount =eligible_amount, # eligible_amount =eligible_amount,
reference = reference, # reference = reference,
created_at=datetime.now(timezone.utc), # created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc) # updated_at=datetime.now(timezone.utc)
) # )
#
try: # try:
db.session.add(loan) # db.session.add(loan)
except IntegrityError as err: # except IntegrityError as err:
raise ValueError(f"Database integrity error: {err}") # raise ValueError(f"Database integrity error: {err}")
return loan # return loan
#
@classmethod # @classmethod
def has_active_loans(cls, customer_id): # def has_active_loans(cls, customer_id):
active_loans = cls.query.filter_by( # active_loans = cls.query.filter_by(
customer_id=customer_id, # customer_id=customer_id,
status='active' # status='active'
).count() # ).count()
#
if active_loans > 0: # if active_loans > 0:
return False # return False
return True # return True
#
#
@classmethod # @classmethod
def get_customer_loan(cls, loan_id, customer_id): # def get_customer_loan(cls, loan_id, customer_id):
""" # """
Get customer's active loans by loan_id. # Get customer's active loans by loan_id.
""" # """
loan = cls.query.filter_by(id = loan_id, customer_id = customer_id).first() # loan = cls.query.filter_by(id = loan_id, customer_id = customer_id).first()
if not loan: # if not loan:
raise ValueError(f"Loan with ID {loan_id} does not exist or does not belong to customer {customer_id}.") # raise ValueError(f"Loan with ID {loan_id} does not exist or does not belong to customer {customer_id}.")
return loan # return loan
#
@classmethod # @classmethod
def get_customer_original_loan(cls, customer_id, original_transaction): # def get_customer_original_loan(cls, customer_id, original_transaction):
""" # """
Get customer's original loan offer. # Get customer's original loan offer.
""" # """
original_loan = cls.query.filter(and_( cls.customer_id ==customer_id, cls.original_transaction==original_transaction, cls.transaction_id==original_transaction )).first() # original_loan = cls.query.filter(and_( cls.customer_id ==customer_id, cls.original_transaction==original_transaction, cls.transaction_id==original_transaction )).first()
if not original_loan: # if not original_loan:
return None # return None
#
logger.info(f" get_customer_original_loan ==>>>> {original_loan}") # logger.info(f" get_customer_original_loan ==>>>> {original_loan}")
return original_loan # return original_loan
#
@classmethod # @classmethod
def get_customer_last_loan(cls, customer_id): # def get_customer_last_loan(cls, customer_id):
""" # """
Get customer's active loans. # Get customer's active loans.
""" # """
logger.info(f"get_customer_last_loan [customer_id] ==>>>> {customer_id}") # logger.info(f"get_customer_last_loan [customer_id] ==>>>> {customer_id}")
# loan = cls.query.filter_by( cls.customer_id == customer_id).first() # # loan = cls.query.filter_by( cls.customer_id == customer_id).first()
loan = cls.query.filter(and_( cls.customer_id ==customer_id, cls.status=='active')).first() # loan = cls.query.filter(and_( cls.customer_id ==customer_id, cls.status=='active')).first()
#
if not loan: # if not loan:
return None # return None
# loan = { # # loan = {
# "original_transaction":"", # # "original_transaction":"",
# "eligible_amount": 0, # # "eligible_amount": 0,
# "loan_amount": 0, # # "loan_amount": 0,
# "customer_id": customer_id, # # "customer_id": customer_id,
# "transaction_id": "", # # "transaction_id": "",
# "resultDescription": "No Active Loan" # # "resultDescription": "No Active Loan"
# } # # }
logger.info(f" get_customer_last_loan ==>>>> {loan}") # logger.info(f" get_customer_last_loan ==>>>> {loan}")
return loan # return loan
#
@classmethod # @classmethod
def get_active_loans_by_original_transaction(cls, original_transaction_id): # def get_active_loans_by_original_transaction(cls, original_transaction_id):
""" # """
Get all active loans with the same original_transaction ID. # Get all active loans with the same original_transaction ID.
""" # """
#
active_loans = cls.query.filter_by( # active_loans = cls.query.filter_by(
original_transaction=original_transaction_id, # original_transaction=original_transaction_id,
# status='active' # # status='active'
).all() # ).all()
#
return active_loans # return active_loans
#
#
@classmethod # @classmethod
def update_status(cls, loan_id, status): # def update_status(cls, loan_id, status):
""" # """
Update the status of the loan with the given loan_id. # Update the status of the loan with the given loan_id.
""" # """
# Retrieve loan # # Retrieve loan
loan = cls.query.get(loan_id) # loan = cls.query.get(loan_id)
#
if not loan: # if not loan:
raise ValueError(f"Loan with ID {loan_id} does not exist.") # raise ValueError(f"Loan with ID {loan_id} does not exist.")
#
if loan.status == status: # if loan.status == status:
return # return
#
# Update loan status and the updated_at timestamp # # Update loan status and the updated_at timestamp
loan.status = status # loan.status = status
#
#
@classmethod # @classmethod
def get_daily_loan_count(cls, customer_id, product_id): # def get_daily_loan_count(cls, customer_id, product_id):
""" # """
Returns the count of loans created today for a customer. # Returns the count of loans created today for a customer.
""" # """
#
start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) # start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = start_of_day + timedelta(days=1) # end_of_day = start_of_day + timedelta(days=1)
#
return cls.query.filter_by( # return cls.query.filter_by(
customer_id=customer_id, # customer_id=customer_id,
product_id=product_id, # product_id=product_id,
).filter( # ).filter(
cls.created_at >= start_of_day, # cls.created_at >= start_of_day,
cls.created_at < end_of_day # cls.created_at < end_of_day
).count() # ).count()
#
#
def to_dict(self): # def to_dict(self):
""" # """
Convert the Loan object to a dictionary format for JSON serialization. # Convert the Loan object to a dictionary format for JSON serialization.
""" # """
return { # return {
'debtId': self.id, # 'debtId': self.id,
'transactionId': self.transaction_id, # 'transactionId': self.transaction_id,
'loanRef': self.reference, # 'loanRef': self.reference,
'productId': self.product_id, # 'productId': self.product_id,
'initialLoanAmount': self.initial_loan_amount, # 'initialLoanAmount': self.initial_loan_amount,
'currentLoanAmount': self.current_loan_amount, # 'currentLoanAmount': self.current_loan_amount,
'defaultPenaltyFee': self.default_penalty_fee, # 'defaultPenaltyFee': self.default_penalty_fee,
'continuousFee': self.continuous_fee, # 'continuousFee': self.continuous_fee,
'collectionType': self.collection_type, # 'collectionType': self.collection_type,
'upfrontFee': self.upfront_fee, # 'upfrontFee': self.upfront_fee,
'repaymentAmount': self.repayment_amount, # 'repaymentAmount': self.repayment_amount,
'installmentAmount': self.installment_amount, # 'installmentAmount': self.installment_amount,
'status': self.status, # 'status': self.status,
'tenor': self.tenor, # 'tenor': self.tenor,
'dueDate': self.due_date.isoformat() if self.due_date else None, # 'dueDate': self.due_date.isoformat() if self.due_date else None,
'loanDate': self.created_at.isoformat() if self.created_at else None, # 'loanDate': self.created_at.isoformat() if self.created_at else None,
} # }
#
def __repr__(self): # def __repr__(self):
return f'<Loan {self.id}>' # return f'<Loan {self.id}>'
app/models/loan_charge.py
+91 -91
View File
@@ -1,91 +1,91 @@
from datetime import datetime, timezone, timedelta # from datetime import datetime, timezone, timedelta
from app.extensions import db # from app.extensions import db
from sqlalchemy.orm import relationship # from sqlalchemy.orm import relationship
from app.utils.logger import logger # from app.utils.logger import logger
from sqlalchemy.sql import func # from sqlalchemy.sql import func
#
#
class LoanCharge(db.Model): # class LoanCharge(db.Model):
__tablename__ = 'loan_charges' # __tablename__ = 'loan_charges'
#
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id = db.Column(db.Integer, primary_key=True, autoincrement=True)
loan_id = db.Column(db.Integer, nullable=False) # loan_id = db.Column(db.Integer, nullable=False)
transaction_id = db.Column(db.String(50), nullable=True) # transaction_id = db.Column(db.String(50), nullable=True)
code = db.Column(db.String(50), nullable=False) # code = db.Column(db.String(50), nullable=False)
amount = db.Column(db.Float, default=0.0) # amount = db.Column(db.Float, default=0.0)
percent = db.Column(db.Float, default=0.0) # percent = db.Column(db.Float, default=0.0)
description = db.Column(db.Text, nullable=True) # description = db.Column(db.Text, nullable=True)
due = db.Column(db.Integer, nullable=False) # due = db.Column(db.Integer, nullable=False)
due_date = db.Column(db.DateTime, nullable=True) # due_date = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime(timezone=True), server_default=func.now()) # created_at = db.Column(db.DateTime(timezone=True), server_default=func.now())
updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) # updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
# loan = relationship( # # loan = relationship(
# "Loan", # # "Loan",
# primaryjoin="LoanCharge.loan_id == Loan.id", # # primaryjoin="LoanCharge.loan_id == Loan.id",
# foreign_keys=[loan_id], # # foreign_keys=[loan_id],
# back_populates="loan_charges", # # back_populates="loan_charges",
# ) # # )
#
@classmethod # @classmethod
def create_charges_for_loan(cls, loan_id, transaction_id, charges, referenced_amount = 0.0): # def create_charges_for_loan(cls, loan_id, transaction_id, charges, referenced_amount = 0.0):
""" # """
Create loan charges for a given loan. # Create loan charges for a given loan.
#
Args: # Args:
loan_id (int): ID of the loan to associate charges with. # loan_id (int): ID of the loan to associate charges with.
charges (list): A list of dictionaries with keys: # charges (list): A list of dictionaries with keys:
code (str), amount (float), percent (float), description (str), due (int) # code (str), amount (float), percent (float), description (str), due (int)
""" # """
# if not charges or not isinstance(charges, list): # # if not charges or not isinstance(charges, list):
# raise ValueError("Charges must be a non-empty list of dictionaries") # # raise ValueError("Charges must be a non-empty list of dictionaries")
#
if loan_id is None: # if loan_id is None:
raise ValueError("loan_id cannot be None") # raise ValueError("loan_id cannot be None")
#
loan_charges = [] # loan_charges = []
now = datetime.now(timezone.utc) # now = datetime.now(timezone.utc)
#
#
subset_keys = ['interest', 'management', 'insurance', 'vat'] # subset_keys = ['interest', 'management', 'insurance', 'vat']
for item in subset_keys: # for item in subset_keys:
charge = charges[item] # charge = charges[item]
due_days = charge['due_days'] # getattr(charge, "due_days", 0) # due_days = charge['due_days'] # getattr(charge, "due_days", 0)
amount = charge['fee'] # getattr(charge, "fee", 0.0) # amount = charge['fee'] # getattr(charge, "fee", 0.0)
percent = charge['rate'] # getattr(charge, "rate", 0.0) # percent = charge['rate'] # getattr(charge, "rate", 0.0)
code = charge['code'] # getattr(charge, "code","") # code = charge['code'] # getattr(charge, "code","")
description = charge['description'] # getattr(charge, "description", "") # description = charge['description'] # getattr(charge, "description", "")
#
charge_obj = cls( # charge_obj = cls(
loan_id = loan_id, # loan_id = loan_id,
transaction_id = transaction_id, # transaction_id = transaction_id,
code = code, # code = code,
amount = round(amount, 2), # amount = round(amount, 2),
percent = percent, # percent = percent,
description = description, # description = description,
due = due_days, # due = due_days,
due_date = now + timedelta(days=due_days), # due_date = now + timedelta(days=due_days),
created_at=datetime.now(timezone.utc), # created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc) # updated_at=datetime.now(timezone.utc)
) # )
#
db.session.add(charge_obj) # db.session.add(charge_obj)
loan_charges.append(charge_obj) # loan_charges.append(charge_obj)
#
return loan_charges # return loan_charges
#
#
#
def to_dict(self): # def to_dict(self):
return { # return {
'id': self.id, # 'id': self.id,
'loanId': self.loan_id, # 'loanId': self.loan_id,
'transactionId': self.transaction_id, # 'transactionId': self.transaction_id,
'code': self.code, # 'code': self.code,
'amount': self.amount, # 'amount': self.amount,
'percent': self.percent, # 'percent': self.percent,
'description': self.description, # 'description': self.description,
'due': self.due, # 'due': self.due,
} # }
#
def __repr__(self): # def __repr__(self):
return f"<LoanCharge {self.id} - Loan {self.loan_id} - {self.code}>" # return f"<LoanCharge {self.id} - Loan {self.loan_id} - {self.code}>"
app/models/loan_repayment_schedule.py
+72 -72
View File
@@ -1,72 +1,72 @@
from datetime import datetime, timezone # from datetime import datetime, timezone
from app.extensions import db # from app.extensions import db
from sqlalchemy.orm import relationship # from sqlalchemy.orm import relationship
from dateutil.relativedelta import relativedelta # from dateutil.relativedelta import relativedelta
from sqlalchemy.sql import func # from sqlalchemy.sql import func
#
class LoanRepaymentSchedule(db.Model): # class LoanRepaymentSchedule(db.Model):
__tablename__ = 'loan_repayment_schedules' # __tablename__ = 'loan_repayment_schedules'
#
id = db.Column(db.Integer, primary_key=True, autoincrement=True) # id = db.Column(db.Integer, primary_key=True, autoincrement=True)
loan_id = db.Column(db.Integer, nullable=False) # loan_id = db.Column(db.Integer, nullable=False)
transaction_id = db.Column(db.String(50), nullable=True) # transaction_id = db.Column(db.String(50), nullable=True)
product_id = db.Column(db.String(20), nullable=True) # product_id = db.Column(db.String(20), nullable=True)
installment_number = db.Column(db.Integer, nullable=False) # installment_number = db.Column(db.Integer, nullable=False)
due_date = db.Column(db.DateTime, nullable=False) # due_date = db.Column(db.DateTime, nullable=False)
installment_amount= db.Column(db.Float, default=0.0) # installment_amount= db.Column(db.Float, default=0.0)
total_repayment_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 = db.Column(db.Boolean, default=False)
paid_at = db.Column(db.DateTime, nullable=True) # paid_at = db.Column(db.DateTime, nullable=True)
#
created_at = db.Column(db.DateTime(timezone=True), server_default=func.now()) # created_at = db.Column(db.DateTime(timezone=True), server_default=func.now())
updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) # updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
# loan = relationship( # # loan = relationship(
# "Loan", # # "Loan",
# primaryjoin="LoanRepaymentSchedule.loan_id == Loan.id", # # primaryjoin="LoanRepaymentSchedule.loan_id == Loan.id",
# foreign_keys=[loan_id], # # foreign_keys=[loan_id],
# back_populates="loan_repayment_schedules", # # back_populates="loan_repayment_schedules",
# ) # # )
#
#
@classmethod # @classmethod
def add_repayment_schedule(cls, loan, num_schedules, transaction_id): # def add_repayment_schedule(cls, loan, num_schedules, transaction_id):
""" # """
Add repayment schedules for a given loan. # Add repayment schedules for a given loan.
""" # """
now = datetime.now(timezone.utc) # now = datetime.now(timezone.utc)
schedules = [] # schedules = []
#
for i in range(num_schedules): # for i in range(num_schedules):
due_date = now + relativedelta(months=i + 1) # due_date = now + relativedelta(months=i + 1)
schedule = LoanRepaymentSchedule( # schedule = LoanRepaymentSchedule(
loan_id=loan.id, # loan_id=loan.id,
installment_number=i + 1, # installment_number=i + 1,
due_date=due_date, # due_date=due_date,
total_repayment_amount = round(loan.repayment_amount, 2), # total_repayment_amount = round(loan.repayment_amount, 2),
installment_amount=round(loan.installment_amount, 2), # installment_amount=round(loan.installment_amount, 2),
product_id = loan.product_id, # product_id = loan.product_id,
transaction_id = transaction_id, # transaction_id = transaction_id,
created_at=datetime.now(timezone.utc), # created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc) # updated_at=datetime.now(timezone.utc)
) # )
#
db.session.add(schedule) # db.session.add(schedule)
schedules.append(schedule) # schedules.append(schedule)
#
return schedules # return schedules
#
def to_dict(self): # def to_dict(self):
return { # return {
'id': self.id, # 'id': self.id,
'loanId': self.loan_id, # 'loanId': self.loan_id,
'installmentNumber': self.installment_number, # 'installmentNumber': self.installment_number,
'dueDate': self.due_date.isoformat(), # 'dueDate': self.due_date.isoformat(),
'principalAmount': self.principal_amount, # 'principalAmount': self.principal_amount,
'interestAmount': self.interest_amount, # 'interestAmount': self.interest_amount,
'totalInstallment': self.total_installment, # 'totalInstallment': self.total_installment,
'paid': self.paid, # 'paid': self.paid,
'paidAt': self.paid_at.isoformat() if self.paid_at else None # 'paidAt': self.paid_at.isoformat() if self.paid_at else None
} # }
#
def __repr__(self): # def __repr__(self):
return f'<LoanRepaymentSchedule Loan:{self.loan_id} Installment:{self.installment_number}>' # return f'<LoanRepaymentSchedule Loan:{self.loan_id} Installment:{self.installment_number}>'
app/models/offer.py
+94 -94
View File
@@ -1,94 +1,94 @@
from datetime import datetime, timezone # from datetime import datetime, timezone
from app.extensions import db # from app.extensions import db
from app.models.charge import Charge # from app.models.charge import Charge
from sqlalchemy.orm import relationship # from sqlalchemy.orm import relationship
from sqlalchemy.sql import func # from sqlalchemy.sql import func
#
class Offer(db.Model): # class Offer(db.Model):
__tablename__ = 'offers' # __tablename__ = 'offers'
#
id = db.Column(db.String, primary_key=True) # id = db.Column(db.String, primary_key=True)
product_id = db.Column(db.String, nullable=False) # product_id = db.Column(db.String, nullable=False)
min_amount = db.Column(db.Float, nullable=False) # min_amount = db.Column(db.Float, nullable=False)
max_amount = db.Column(db.Float, nullable=False) # max_amount = db.Column(db.Float, nullable=False)
tenor = db.Column(db.Integer, nullable=False) # tenor = db.Column(db.Integer, nullable=False)
schedule = db.Column(db.Integer, nullable=True) # schedule = db.Column(db.Integer, nullable=True)
interest_rate = db.Column(db.Float, default=3.0) # interest_rate = db.Column(db.Float, default=3.0)
management_rate = db.Column(db.Float, default=1.0) # management_rate = db.Column(db.Float, default=1.0)
insurance_rate = db.Column(db.Float, default=1.0) # insurance_rate = db.Column(db.Float, default=1.0)
vat_rate = db.Column(db.Float, default=7.5) # vat_rate = db.Column(db.Float, default=7.5)
list_order = db.Column(db.Integer, nullable=True) # list_order = db.Column(db.Integer, nullable=True)
max_daily_loans = db.Column(db.Integer, nullable=True) # max_daily_loans = db.Column(db.Integer, nullable=True)
max_active_loans = db.Column(db.Integer, nullable=True) # max_active_loans = db.Column(db.Integer, nullable=True)
max_life_loans = db.Column(db.Integer, nullable=True) # max_life_loans = db.Column(db.Integer, nullable=True)
created_at = db.Column(db.DateTime(timezone=True), server_default=func.now()) # created_at = db.Column(db.DateTime(timezone=True), server_default=func.now())
updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) # updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
#
# charges = relationship( # # charges = relationship(
# "Charge", # # "Charge",
# primaryjoin="Offer.id == Charge.offer_id", # # primaryjoin="Offer.id == Charge.offer_id",
# foreign_keys="Charge.offer_id", # # foreign_keys="Charge.offer_id",
# back_populates="offer", # # back_populates="offer",
# ) # # )
#
@classmethod # @classmethod
def get_all_offers(cls): # def get_all_offers(cls):
""" # """
Return all offers in dictionary format. # Return all offers in dictionary format.
""" # """
offers = cls.query.all() # offers = cls.query.all()
#
if not offers: # if not offers:
raise ValueError(f"No available offers") # raise ValueError(f"No available offers")
return offers # return offers
#
@classmethod # @classmethod
def is_valid_offer(cls, offer_id): # def is_valid_offer(cls, offer_id):
offer = cls.query.filter_by(id=str(offer_id)).first() # offer = cls.query.filter_by(id=str(offer_id)).first()
#
#
if not offer: # if not offer:
return False # return False
return offer # return offer
#
@classmethod # @classmethod
def get_offer_by_id(cls, offer_id): # def get_offer_by_id(cls, offer_id):
""" # """
Return an offer by its ID. # Return an offer by its ID.
""" # """
offer = cls.query.filter_by(id=str(offer_id)).first() # offer = cls.query.filter_by(id=str(offer_id)).first()
#
if not offer: # if not offer:
raise ValueError(f"Offer with ID {offer_id} not found") # raise ValueError(f"Offer with ID {offer_id} not found")
return offer # return offer
#
@classmethod # @classmethod
def get_offer_by_product_id(cls, product_id): # def get_offer_by_product_id(cls, product_id):
""" # """
Return an offer by its product ID. # Return an offer by its product ID.
""" # """
offer = cls.query.filter_by(product_id=str(product_id)).first() # offer = cls.query.filter_by(product_id=str(product_id)).first()
#
if not offer: # if not offer:
raise ValueError(f"Offer with Product ID {product_id} not found") # raise ValueError(f"Offer with Product ID {product_id} not found")
return offer # return offer
#
def to_dict(self): # def to_dict(self):
return { # return {
"offerId": self.id, # "offerId": self.id,
"productId": self.product_id, # "productId": self.product_id,
"minAmount": self.min_amount, # "minAmount": self.min_amount,
"maxAmount": self.max_amount, # "maxAmount": self.max_amount,
"tenor": self.tenor, # "tenor": self.tenor,
"interest_rate": self.interest_rate, # "interest_rate": self.interest_rate,
"management_rate": self.management_rate, # "management_rate": self.management_rate,
"insurance_rate": self.insurance_rate, # "insurance_rate": self.insurance_rate,
"vat_rate": self.vat_rate, # "vat_rate": self.vat_rate,
"maxDailyLoans": self.max_daily_loans, # "maxDailyLoans": self.max_daily_loans,
"maxActiveLoans": self.max_active_loans, # "maxActiveLoans": self.max_active_loans,
"maxLifeLoans": self.max_life_loans # "maxLifeLoans": self.max_life_loans
#
} # }
#
def __repr__(self): # def __repr__(self):
return f'<LoanOffer {self.id}>' # return f'<LoanOffer {self.id}>'