TEST/digifi-flaskA002
2
0
Fork 0
Code Issues Pull Requests Releases Activity

bluprints, schemas,helpers and routes

Browse Source
This commit is contained in:
VivianDee
2025-03-20 17:46:34 +01:00
commit aafa9992e3
55 changed files with 1847 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__init__.py
+29
View File
@@ -0,0 +1,29 @@
from flask import Flask
# from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_cors import CORS
from app.config import Config
from app.routes import api
# Initialize extensions
# db = SQLAlchemy()
ma = Marshmallow()
def create_app():
""" Factory function to create a Flask app instance """
app = Flask(__name__)
# Load configuration
app.config.from_object(Config)
# Initialize extensions
# db.init_app(app)
ma.init_app(app)
CORS(app)
# Register blueprints
app.register_blueprint(api, url_prefix="/api")
return app
app/blueprints/__init__.py
+18
View File
@@ -0,0 +1,18 @@
from app.blueprints.eligibility_check import EligibilityCheckService
from app.blueprints.select_offer import SelectOfferService
from app.blueprints.provide_loan import ProvideLoanService
from app.blueprints.loan_information import LoanInformationService
from app.blueprints.repayment import RepaymentService
from app.blueprints.customer_consent import CustomerConsentService
from app.blueprints.notification_callback import NotificationCallbackService
from app.blueprints.rac_check import RACCheckService
from app.blueprints.disbursement import DisbursementService
from app.blueprints.collect_loan import CollectLoanService
from app.blueprints.transaction_verify import TransactionVerifyService
from app.blueprints.penal_charge import PenalChargeService
from app.blueprints.revoke_enable_consent import RevokeEnableConsentService
from app.blueprints.token_validation import TokenValidationService
from app.blueprints.lien_check import LienCheckService
from app.blueprints.new_transaction_check import NewTransactionCheckService
from app.blueprints.sms import SMSService
from app.blueprints.bulk_sms import BulkSMSService
app/blueprints/bulk_sms.py
+55
View File
@@ -0,0 +1,55 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.bulk_sms import BulkSMSSchema
class BulkSMSService:
@staticmethod
def process_request(data):
"""
Process the Bulk SMS request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing BulkSMS request")
# Validate input data using BulkSMSSchema
schema = BulkSMSSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated Bulk SMS sending logic
response_data = {
"data": "",
"statusCode": 200,
"isSuccessful": True,
"errorMessage": None
}
# return ResponseHelper.success(
# data=response_data,
# message="Bulk SMS sent successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/collect_loan.py
+63
View File
@@ -0,0 +1,63 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.collect_loan import CollectLoanSchema
class CollectLoanService:
@staticmethod
def process_request(data):
"""
Process the CollectLoan request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing CollectLoan request")
# Validate input data using CollectLoanSchema
schema = CollectLoanSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"transactionId": "T002",
"debtId": "273194670",
"customerId": "CN621868",
"accountId": "2017821799",
"productId": "101",
"collectAmount": 60000.00,
"penalCharge": 0,
"lienAmount": 20000,
"countryId": "01",
"comment": "Testing CollectionLoanRequest",
"resultCode": "00",
"resultDescription": "Loan Collection Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Loan collection completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/customer_consent.py
+55
View File
@@ -0,0 +1,55 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.customer_consent import CustomerConsentSchema
class CustomerConsentService:
@staticmethod
def process_request(data):
"""
Process the CustomerConsent request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing CustomerConsent request")
# Validate input data using the CustomerConsent schema
schema = CustomerConsentSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"resultCode": "00",
"resultDescription": "Request is received"
}
# return ResponseHelper.success(
# data=response_data,
# message="Customer consent processed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/disbursement.py
+55
View File
@@ -0,0 +1,55 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.disbursement import DisbursementSchema
class DisbursementService:
@staticmethod
def process_request(data):
"""
Process the Disbursement request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing Disbursement request")
# Validate input data using DisbursementSchema
schema = DisbursementSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"disbursement_id": validated_data.get("disbursement_id", "11223"),
"status": "Completed",
"amount": validated_data.get("amount"),
"recipient_account": validated_data.get("recipient_account"),
}
# return ResponseHelper.success(
# data=response_data,
# message="Disbursement completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/eligibility_check.py
+77
View File
@@ -0,0 +1,77 @@
from flask import session
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.eligibility_check import EligibilityCheckSchema
from marshmallow import ValidationError
class EligibilityCheckService:
@staticmethod
def process_request(data):
"""
Process the EligibilityCheck request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing EligibilityCheck request")
# Validate input data using Schema
schema = EligibilityCheckSchema()
validated_data = schema.load(data) # Raises an error if invalid
# Example: Validate data, perform calculations, etc.
if not data:
return ResponseHelper.error("Invalid input data", status_code=400)
# Simulate processing
response_data = {
"customerId": "CN621868",
"transactionId": "Tr201712RK9232P115",
"msisdn": "3451342",
"eligibleOffers": [
{
"minamount": 5000,
"maxamount": 20000,
"productId": 101,
"offerid": 101,
"Tenor": 30
},
{
"minamount": 20000,
"maxamount": 50000,
"productId": 102,
"offerid": 102,
"Tenor": 60
}
],
"resultCode": "00",
"resultDescription": "Successful"
}
# Return a success response
# return ResponseHelper.success(
# data=response_data,
# message="Eligibility check completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred during EligibilityCheck processing: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/lien_check.py
+54
View File
@@ -0,0 +1,54 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.lien_check import LienCheckSchema
class LienCheckService:
@staticmethod
def process_request(data):
"""
Process the LienCheck request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing LienCheck request")
# Validate input data using LienCheckSchema
schema = LienCheckSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated lien check logic
response_data = {
"lienAmount": 20000.0,
"resultCode": "00",
"resultDescription": "Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Lien check completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/loan_information.py
+67
View File
@@ -0,0 +1,67 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.loan_information import LoanInformationSchema
class LoanInformationService:
@staticmethod
def process_request(data):
"""
Process the Loan Information request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing LoanInformation request")
# Validate input data using the imported schema
schema = LoanInformationSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"customerId": "CN621868",
"loans": [
{
"debtId": "123456789",
"loanDate": "2019-10-18 14:26:21.063",
"dueDate": "2019-11-20 14:26:21.063",
"currentLoanAmount": 8500.0,
"initialLoanAmount": 10000.0,
"defaultFee": 0.0,
"continuousFee": 0.0,
"productId": "101"
}
],
"resultCode": "00",
"resultDescription": "Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Loan information retrieved successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/new_transaction_check.py
+61
View File
@@ -0,0 +1,61 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.new_transaction_check import NewTransactionCheckSchema
class NewTransactionCheckService:
@staticmethod
def process_request(data):
"""
Process the NewTransactionCheck request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing NewTransactionCheck request")
# Validate input data using NewTransactionCheckSchema
schema = NewTransactionCheckSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated transaction check logic
response_data = {
"transactionId": "24110114545374721",
"data": {
"transactionId": "241101",
"providedAmount": 1000.00,
"collectedAmount": 0.00,
"resultCode": "00",
"resultDescription": "Loan Provision is successful"
},
"resultCode": "00",
"resultDescription": "SUCCESS"
}
# return ResponseHelper.success(
# data=response_data,
# message="New transaction check completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/notification_callback.py
+54
View File
@@ -0,0 +1,54 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.notification_callback import NotificationCallbackSchema
class NotificationCallbackService:
@staticmethod
def process_request(data):
"""
Process the NotificationCallback request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing NotificationCallback request")
# Validate input data using the NotificationCallback schema
schema = NotificationCallbackSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"resultCode": "00",
"resultDescription": "Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Notification callback processed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/penal_charge.py
+55
View File
@@ -0,0 +1,55 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.penal_charge import PenalChargeSchema
class PenalChargeService:
@staticmethod
def process_request(data):
"""
Process the PenalCharge request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing PenalCharge request")
# Validate input data using PenalChargeSchema
schema = PenalChargeSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"resultCode": "00",
"resultDescription": "Penal charge debited successfully"
}
# return ResponseHelper.success(
# data=response_data,
# message="Penal charge applied successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/provide_loan.py
+59
View File
@@ -0,0 +1,59 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.provide_loan import ProvideLoanSchema
class ProvideLoanService:
@staticmethod
def process_request(data):
"""
Process the ProvideLoan request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing ProvideLoan request")
# Validate input data using the imported schema
schema = ProvideLoanSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Business logic - providing a loan
response_data ={
"requestId": "202111170001371256908",
"transactionId": "Tr201712RK9232P115",
"customerId": "CN621868",
"accountId": "ACN8263457",
"msisdn": "3451342",
"resultCode": "00",
"resultDescription": "Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Loan successfully provided"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/rac_check.py
+67
View File
@@ -0,0 +1,67 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.rac_check import RACCheckSchema
class RACCheckService:
@staticmethod
def process_request(data):
"""
Process the RACCheck request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing RACCheck request")
# Validate input data using RACCheckSchema
schema = RACCheckSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"resultCode": "00",
"RACResponse": {
"SalaryAccount": "1",
"BVN": "1",
"BVNAttachedToAccount": "1",
"CRMS": "1",
"CRC": "1",
"AccountStatus": "1",
"Lien": "1",
"NoBouncedCheck": "1",
"Whitelist": "1",
"NoPastDueSalaryLoan": "1",
"NoPastDueOtherLoan": "1"
},
"resultDescription": "RAC Check Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="RAC check completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/repayment.py
+54
View File
@@ -0,0 +1,54 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.repayment import RepaymentSchema
class RepaymentService:
@staticmethod
def process_request(data):
"""
Process the Repayment request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing Repayment request")
# Validate input data using the Repayment schema
schema = RepaymentSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"repayment_id": "67890",
"status": "Paid",
"amount": validated_data.get("amount", 0), # Example: Use validated field
}
# return ResponseHelper.success(
# data=response_data,
# message="Repayment processed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/revoke_enable_consent.py
+58
View File
@@ -0,0 +1,58 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.revoke_enable_consent import RevokeEnableConsentSchema
class RevokeEnableConsentService:
@staticmethod
def process_request(data):
"""
Process the RevokeEnableConsent request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing RevokeEnableConsent request")
# Validate input data using RevokeEnableConsentSchema
schema = RevokeEnableConsentSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"$type": "RevokeEnableConsentResponse",
"customerId": "CN621868",
"accountId": "2017821799",
"resultCode": "00",
"resultDescription": "Success"
}
# return ResponseHelper.success(
# data=response_data,
# message="Consent revocation processed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/select_offer.py
+109
View File
@@ -0,0 +1,109 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.select_offer import SelectOfferSchema
class SelectOfferService:
@staticmethod
def process_request(data):
"""
Process the SelectOffer request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing SelectOffer request")
# Validate input data using the imported schema
schema = SelectOfferSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Business logic - selecting an offer
response_data = {
"outstandingDebtAmount": 0,
"requestId": "202111170001371256908",
"transactionId": "1231231321232",
"customerId": "1256907",
"accountId": "5948306019",
"offers": [
{
"offerId": "14451",
"productId": "2030",
"amount": 10000.0,
"upfrontPayment": 1000.0,
"interestRate": 3.0,
"managementRate": 1.0,
"managementFee": 1.0,
"insuranceRate": 1.0,
"insuranceFee": 100.0,
"vatRate": 7.5,
"vatAmount": 100.0,
"recommendedRepaymentDates": ["2022-11-30"],
"installmentAmount": 11000.0,
"totalRepaymentAmount": 11000.0
},
{
"offerId": "16645",
"productId": "2060",
"amount": 10000.0,
"upfrontPayment": 0.0,
"interestRate": 3.0,
"managementRate": 1.0,
"managementFee": 1.0,
"insuranceRate": 1.0,
"insuranceFee": 100.0,
"vatRate": 7.5,
"vatAmount": 100.0,
"recommendedRepaymentDates": ["2022-11-30", "2023-12-30"],
"installmentAmount": 5761.9,
"totalRepaymentAmount": 11523.8
},
{
"offerId": "122212",
"productId": "2090",
"amount": 10000.0,
"upfrontPayment": 0.0,
"interestRate": 10.0,
"managementRate": 1.0,
"managementFee": 1.0,
"insuranceRate": 1.0,
"insuranceFee": 100.0,
"vatRate": 7.5,
"vatAmount": 100.0,
"recommendedRepaymentDates": ["2022-11-30", "2022-12-30", "2023-01-29"],
"installmentAmount": 4021.15,
"totalRepaymentAmount": 12063.45
}
],
"resultCode": "00",
"resultDescription": "Successful"
}
# return ResponseHelper.success(
# data=response_data,
# message="Offer selection completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/sms.py
+56
View File
@@ -0,0 +1,56 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.sms import SMSSchema
class SMSService:
@staticmethod
def process_request(data):
"""
Process the SMS request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing SMS request")
# Validate input data using SMSSchema
schema = SMSSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated SMS sending logic
response_data = {
"data": "",
"statusCode": 200,
"isSuccessful": True,
"errorMessage": None
}
# return ResponseHelper.success(
# data=response_data,
# message="SMS sent successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/token_validation.py
+58
View File
@@ -0,0 +1,58 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.token_validation import TokenValidationSchema
class TokenValidationService:
@staticmethod
def process_request(data):
"""
Process the TokenValidation request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing TokenValidation request")
# Validate input data using TokenValidationSchema
schema = TokenValidationSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated token validation logic
response_data = {
"Authenticated": True,
"AuthenticatedMessage": "The user Oluwole Olusoga has successfully authenticated!",
"ResponseCode": "00",
"ResponseMessage": "Successful",
"RequestId": "SMB1234567"
}
# return ResponseHelper.success(
# data=response_data,
# message="Token validation completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/blueprints/transaction_verify.py
+61
View File
@@ -0,0 +1,61 @@
from flask import request
from marshmallow import ValidationError
from app.utils.logger import logger
from app.helpers.response_helper import ResponseHelper
from app.schemas.transaction_verify import TransactionVerifySchema
class TransactionVerifyService:
@staticmethod
def process_request(data):
"""
Process the TransactionVerify request.
Args:
data (dict): The request data.
Returns:
dict: A standardized response.
"""
try:
logger.info("Processing TransactionVerify request")
# Validate input data using TransactionVerifySchema
schema = TransactionVerifySchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
# Simulated processing logic
response_data = {
"$type": "TransactionCheckResponse",
"nativeId": "FBN20191031104405CN621868",
"customerId": "CN621868",
"accountId": "2017821799",
"providedAmount": 0.0,
"collectedAmount": 7.50,
"resultCode": "00",
"resultDescription": "Collect Status retrieved successfully."
}
# return ResponseHelper.success(
# data=response_data,
# message="Transaction verification completed successfully"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
return ResponseHelper.error(
message="Invalid input data",
status_code=400,
error=err.messages
)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return ResponseHelper.error(
message="An internal error occurred",
status_code=500,
error=str(e)
)
app/config.py
+10
View File
@@ -0,0 +1,10 @@
import os
class Config:
"""Base configuration for Flask app"""
# SQLALCHEMY_DATABASE_URI = "mysql://root:password@localhost/flask_app"
# SQLALCHEMY_TRACK_MODIFICATIONS = False
# SECRET_KEY = os.environ.get("SECRET_KEY", "your_secret_key")
DEBUG = True
app/helpers/response_helper.py
+213
View File
@@ -0,0 +1,213 @@
from flask import jsonify
from typing import List, Dict, Union, Optional, Any
class ResponseHelper:
"""
A helper class for building standardized JSON responses in Flask.
"""
@staticmethod
def build_response(
status: bool,
message: str,
data: Optional[Union[Dict, List, str]] = None,
status_code: int = 200,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Build a standardized JSON response.
Args:
status (bool): Indicates whether the request was successful.
message (str): A message describing the result of the request.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
status_code (int): The HTTP status code for the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
response = {
"status": status,
"statusCode": status_code,
"message": message,
"data": data if data is not None else {},
"error": error if error is not None else {},
}
return response
@staticmethod
def success(
data: Optional[Union[Dict, List, str]] = None,
message: str = "Successful",
status_code: int = 200,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a success response.
Args:
data (Optional[Union[Dict, List, str]]): The data to return in the response.
message (str): A message describing the result of the request.
status_code (int): The HTTP status code for the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(True, message, data, status_code, error)
@staticmethod
def error(
message: str = "An error occurred",
status_code: int = 400,
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return an error response.
Args:
message (str): A message describing the error.
status_code (int): The HTTP status code for the response.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, status_code, error)
@staticmethod
def created(
data: Optional[Union[Dict, List, str]] = None,
message: str = "Resource created successfully",
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for a created resource.
Args:
data (Optional[Union[Dict, List, str]]): The data to return in the response.
message (str): A message describing the result of the request.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(True, message, data, 201, error)
@staticmethod
def updated(
data: Optional[Union[Dict, List, str]] = None,
message: str = "Resource updated successfully",
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for an updated resource.
Args:
data (Optional[Union[Dict, List, str]]): The data to return in the response.
message (str): A message describing the result of the request.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(True, message, data, 200, error)
@staticmethod
def internal_server_error(
message: str = "Internal Server Error",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for an internal server error.
Args:
message (str): A message describing the error.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, 500, error)
@staticmethod
def unauthorized(
message: str = "Unauthorized",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for an unauthorized request.
Args:
message (str): A message describing the error.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, 401, error)
@staticmethod
def forbidden(
message: str = "Forbidden",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for a forbidden request.
Args:
message (str): A message describing the error.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, 403, error)
@staticmethod
def not_found(
message: str = "Resource not found",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for a not found resource.
Args:
message (str): A message describing the error.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, 404, error)
@staticmethod
def unprocessable_entity(
message: str = "Unprocessable entity",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for an unprocessable entity.
Args:
message (str): A message describing the error.
data (Optional[Union[Dict, List, str]]): The data to return in the response.
error (Optional[Union[Dict, str]]): Any error details to include in the response.
Returns:
Dict[str, Any]: A dictionary representing the JSON response.
"""
return ResponseHelper.build_response(False, message, data, 422, error)
app/middlewares/__init__.py
View File
app/middlewares/cors.py
+9
View File
@@ -0,0 +1,9 @@
# app/middlewares/cors.py
from flask import request
def cors_headers(response):
"""Allow cross-origin requests"""
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, PATCH, DELETE"
response.headers["Access-Control-Allow-Headers"] = "Authorization, Content-Type"
return response
app/middlewares/encryption.py
+14
View File
@@ -0,0 +1,14 @@
# app/middlewares/encryption.py
from cryptography.fernet import Fernet
import os
ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY", Fernet.generate_key())
cipher = Fernet(ENCRYPTION_KEY)
def encrypt_data(data):
"""Encrypt sensitive data"""
return cipher.encrypt(data.encode()).decode()
def decrypt_data(data):
"""Decrypt sensitive data"""
return cipher.decrypt(data.encode()).decode()
app/middlewares/request_validator.py
+11
View File
@@ -0,0 +1,11 @@
# app/middlewares/request_validator.py
from flask import request
from app.helpers.response_helper import ResponseHelper
def validate_json():
"""Ensure request has valid JSON"""
if not request.is_json:
return ResponseHelper.error(
message="Request must be JSON",
status_code=415
)
app/middlewares/verify_api_key.py
+8
View File
@@ -0,0 +1,8 @@
# app/middlewares/auth.py
from flask import request, jsonify
def require_api_key():
"""Middleware to check if API key is present"""
api_key = request.headers.get("X-API-KEY")
if not api_key:
return jsonify({"error": "Missing API key"}), 403
app/routes/__init__.py
+1
View File
@@ -0,0 +1 @@
from .routes import api
app/routes/__pycache__/__init__.cpython-312.pyc
BIN
View File
Binary file not shown.
app/routes/__pycache__/main.cpython-312.pyc
BIN
View File
Binary file not shown.
app/routes/routes.py
+175
View File
@@ -0,0 +1,175 @@
from flask import Blueprint, request, jsonify
from app.blueprints import (
EligibilityCheckService,
SelectOfferService,
ProvideLoanService,
LoanInformationService,
RepaymentService,
CustomerConsentService,
NotificationCallbackService,
RACCheckService,
DisbursementService,
CollectLoanService,
TransactionVerifyService,
PenalChargeService,
RevokeEnableConsentService,
TokenValidationService,
LienCheckService,
NewTransactionCheckService,
SMSService,
BulkSMSService
)
from app.utils.logger import logger
api = Blueprint("api", __name__)
# EligibilityCheck Endpoint
@api.route('/EligibilityCheck', methods=['POST'])
def eligibility_check():
data = request.get_json()
# logger.info(f"EligibilityCheck request received: {data}")
response = EligibilityCheckService.process_request(data)
return jsonify(response)
# SelectOffer Endpoint
@api.route('/SelectOffer', methods=['POST'])
def select_offer():
data = request.get_json()
# logger.info(f"SelectOffer request received: {data}")
response = SelectOfferService.process_request(data)
return jsonify(response)
# ProvideLoan Endpoint
@api.route('/ProvideLoan', methods=['POST'])
def provide_loan():
data = request.get_json()
# logger.info(f"ProvideLoan request received: {data}")
response = ProvideLoanService.process_request(data)
return jsonify(response)
# LoanInformation Endpoint
@api.route('/LoanInformation', methods=['GET'])
def loan_information():
data = request.args.to_dict()
# logger.info(f"LoanInformation request received: {data}")
response = LoanInformationService.process_request(data)
return jsonify(response)
# Repayment Endpoint
@api.route('/Repayment', methods=['POST'])
def repayment():
data = request.get_json()
# logger.info(f"Repayment request received: {data}")
response = RepaymentService.process_request(data)
return jsonify(response)
# CustomerConsent Endpoint
@api.route('/CustomerConsent', methods=['POST'])
def customer_consent():
data = request.get_json()
# logger.info(f"CustomerConsent request received: {data}")
response = CustomerConsentService.process_request(data)
return jsonify(response)
# NotificationCallback Endpoint
@api.route('/NotificationCallback', methods=['POST'])
def notification_callback():
data = request.get_json()
# logger.info(f"NotificationCallback request received: {data}")
response = NotificationCallbackService.process_request(data)
return jsonify(response)
# RACCheck Endpoint
@api.route('/RACCheck', methods=['POST'])
def rac_check():
data = request.get_json()
# logger.info(f"RACCheck request received: {data}")
response = RACCheckService.process_request(data)
return jsonify(response)
# Disbursement Endpoint
@api.route('/Disbursement', methods=['POST'])
def disbursement():
data = request.get_json()
# logger.info(f"Disbursement request received: {data}")
response = DisbursementService.process_request(data)
return jsonify(response)
# CollectLoan Endpoint
@api.route('/CollectLoan', methods=['POST'])
def collect_loan():
data = request.get_json()
# logger.info(f"CollectLoan request received: {data}")
response = CollectLoanService.process_request(data)
return jsonify(response)
# TransactionVerify Endpoint
@api.route('/TransactionVerify', methods=['POST'])
def transaction_verify():
data = request.get_json()
# logger.info(f"TransactionVerify request received: {data}")
response = TransactionVerifyService.process_request(data)
return jsonify(response)
# PenalCharge Endpoint
@api.route('/PenalCharge', methods=['POST'])
def penal_charge():
data = request.get_json()
# logger.info(f"PenalCharge request received: {data}")
response = PenalChargeService.process_request(data)
return jsonify(response)
# RevokeEnableConsent Endpoint
@api.route('/RevokeEnableConsent', methods=['POST'])
def revoke_enable_consent():
data = request.get_json()
# logger.info(f"RevokeEnableConsent request received: {data}")
response = RevokeEnableConsentService.process_request(data)
return jsonify(response)
# TokenValidation Endpoint
@api.route('/TokenValidation', methods=['POST'])
def token_validation():
data = request.get_json()
# logger.info(f"TokenValidation request received: {data}")
response = TokenValidationService.process_request(data)
return jsonify(response)
# LienCheck Endpoint
@api.route('/LienCheck', methods=['POST'])
def lien_check():
data = request.get_json()
# logger.info(f"LienCheck request received: {data}")
response = LienCheckService.process_request(data)
return jsonify(response)
# NewTransactionCheck Endpoint
@api.route('/NewTransactionCheck', methods=['POST'])
def new_transaction_check():
data = request.get_json()
# logger.info(f"NewTransactionCheck request received: {data}")
response = NewTransactionCheckService.process_request(data)
return jsonify(response)
# SMS Endpoint
@api.route('/SMS', methods=['POST'])
def sms():
data = request.get_json()
# logger.info(f"SMS request received: {data}")
response = SMSService.process_request(data)
return jsonify(response)
# BulkSMS Endpoint
@api.route('/BulkSMS', methods=['POST'])
def bulk_sms():
data = request.get_json()
# logger.info(f"BulkSMS request received: {data}")
response = BulkSMSService.process_request(data)
return jsonify(response)
# Health Check Endpoint
@api.route('/health', methods=['GET'])
def health_check():
return {"status": "ok"} , 200
app/schemas/__init__.py
View File
app/schemas/bulk_sms.py
+6
View File
@@ -0,0 +1,6 @@
from marshmallow import Schema, fields
from .sms import SMSSchema
# Bulk SMS Schema
class BulkSMSSchema(Schema):
requests = fields.List(fields.Nested(SMSSchema), required=True)
app/schemas/collect_loan.py
+16
View File
@@ -0,0 +1,16 @@
from marshmallow import Schema, fields
# Collect Loan Schema
class CollectLoanSchema(Schema):
transactionId = fields.Str(required=True)
fbnTransactionId = fields.Str(required=True)
debtId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
productId = fields.Str(required=True)
collectAmount = fields.Float(required=True)
penalCharge = fields.Float(required=False) # Optional
collectionMethod = fields.Int(required=True)
lienAmount = fields.Float(required=True)
countryId = fields.Str(required=True)
comment = fields.Str(required=False) # Optional
app/schemas/customer_consent.py
+11
View File
@@ -0,0 +1,11 @@
from marshmallow import Schema, fields
# Customer Consent Schema
class CustomerConsentSchema(Schema):
type = fields.Str(data_key="$type", required=True)
transactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
requestTime = fields.DateTime(required=True, format="%Y-%m-%d %H:%M:%S.%f")
consentType = fields.Str(required=True)
channel = fields.Str(required=True)
app/schemas/disbursement.py
+17
View File
@@ -0,0 +1,17 @@
from marshmallow import Schema, fields
# Disbursement Schema
class DisbursementSchema(Schema):
requestId = fields.Str(required=True, data_key="requestId")
debtId = fields.Str(required=True, data_key="debtId")
transactionId = fields.Str(required=True, data_key="transactionId")
customerId = fields.Str(required=True, data_key="customerId")
accountId = fields.Str(required=True, data_key="accountId")
productId = fields.Str(required=True, data_key="productId")
provideAmount = fields.Float(required=True, data_key="provideAmount")
collectAmountInterest = fields.Float(required=False, data_key="collectAmountInterest") # Optional
collectAmountMgtFee = fields.Float(required=True, data_key="collectAmountMgtFee")
collectAmountInsurance = fields.Float(required=True, data_key="collectAmountInsurance")
collectAmountVAT = fields.Float(required=True, data_key="collectAmountVAT")
countryId = fields.Str(required=True, data_key="countryId")
comment = fields.Str(required=False, data_key="comment") # Optional
app/schemas/eligibility_check.py
+11
View File
@@ -0,0 +1,11 @@
from marshmallow import Schema, fields
class EligibilityCheckSchema(Schema):
type = fields.Str(data_key="$type", required=True, description="Request type")
transactionId = fields.Str(data_key="transactionId", required=True, description="Transaction ID")
countryCode = fields.Str(data_key="countryCode", required=True, description="Country code (ISO)")
customerId = fields.Str(data_key="customerId", required=True, description="Customer ID")
accountId = fields.Str(data_key="accountId", required=True, description="Account ID")
msisdn = fields.Str(required=True, description="Mobile number")
lienAmount = fields.Float(required=True, description="Amount for lien")
channel = fields.Str(required=True, description="Transaction channel (USSD, Mobile, Web)")
app/schemas/lien_check.py
+8
View File
@@ -0,0 +1,8 @@
from marshmallow import Schema, fields
# Lien Check Schema
class LienCheckSchema(Schema):
transactionId = fields.Str(required=True, metadata={"description": "Unique Identifier in Simbrella system"})
customerId = fields.Str(required=True, metadata={"description": "Unique identifier of customer"})
accountId = fields.Str(required=True, metadata={"description": "Unique identifier of account"})
countryId = fields.Str(required=True, metadata={"description": 'Set to static value "01"'})
app/schemas/loan_information.py
+5
View File
@@ -0,0 +1,5 @@
from marshmallow import Schema, fields
# Loan Information Schema
class LoanInformationSchema(Schema):
loan_id = fields.Str(required=True)
app/schemas/new_transaction_check.py
+12
View File
@@ -0,0 +1,12 @@
from marshmallow import Schema, fields
# New Transaction Check Schema
class NewTransactionCheckSchema(Schema):
transactionId = fields.Str(required=True)
debtId = fields.Str(required=True)
transactionType = fields.Str(required=True, metadata={
"allowed_values": ["Disbursement", "Collection", "PenalCharge"]
})
fbnTransactionId = fields.Str(required=True)
origTransactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
app/schemas/notification_callback.py
+14
View File
@@ -0,0 +1,14 @@
from marshmallow import Schema, fields
# Notification Callback Schema
class NotificationCallbackSchema(Schema):
fbnTransactionId = fields.Str(required=True)
transactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
debtId = fields.Str(required=True)
transactionType = fields.Str(required=True)
amountProvided = fields.Float(required=True)
amountCollected = fields.Float(required=True)
responseCode = fields.Str(required=True)
responseDescription = fields.Str(required=True)
app/schemas/penal_charge.py
+14
View File
@@ -0,0 +1,14 @@
from marshmallow import Schema, fields
# Penal Charge Schema
class PenalChargeSchema(Schema):
transactionId = fields.Str(required=True, metadata={"description": "Unique identifier of transaction in Simbrella system"})
fbnTransactionId = fields.Str(required=True, metadata={"description": "Unique id of the transaction received from FBN in Eligibility or Provision requests"})
debtId = fields.Str(required=True, metadata={"description": "Unique identifier of providing loan in Simbrella system"})
customerId = fields.Str(required=True, metadata={"description": "Unique identifier of a user"})
accountId = fields.Str(required=True, metadata={"description": "Specific identifier of a user’s account"})
penalCharge = fields.Decimal(required=True, metadata={"description": "Penalty amount that needs to be collected from user’s account"})
lienAmount = fields.Decimal(required=True, metadata={"description": "Aggregated (summed up) lien amount"})
countryId = fields.Str(required=True, metadata={"description": 'Set to static value "01"'})
comment = fields.Str(required=False, metadata={"description": "Any additional comment for provided loan operation"})
app/schemas/provide_loan.py
+16
View File
@@ -0,0 +1,16 @@
from marshmallow import Schema, fields
# Provide Loan Schema
class ProvideLoanSchema(Schema):
type = fields.Str(required=True, data_key="$type")
request_id = fields.Str(required=True, data_key="requestId")
transaction_id = fields.Str(required=True, data_key="transactionId")
customer_id = fields.Str(required=True, data_key="customerId")
account_id = fields.Str(required=True, data_key="accountId")
msisdn = fields.Str(required=False, data_key="msisdn")
product_id = fields.Str(required=True, data_key="productId")
lien_amount = fields.Float(required=True, data_key="lienAmount")
requested_amount = fields.Float(required=True, data_key="requestedAmount")
collection_type = fields.Int(required=True, data_key="collectionType")
loan_type = fields.Int(required=True, data_key="loanType")
channel = fields.Str(required=True, data_key="channel")
app/schemas/rac_check.py
+23
View File
@@ -0,0 +1,23 @@
from marshmallow import Schema, fields
class RACItemSchema(Schema):
salaryAccount = fields.Bool(required=True)
bvn = fields.Str(required=True)
crc = fields.Bool(required=True)
crms = fields.Bool(required=True)
accountStatus = fields.Str(required=True)
lien = fields.Bool(required=True)
noBouncedCheck = fields.Bool(required=True)
existingLoan = fields.Bool(required=True)
whitelist = fields.Bool(required=True)
noPastDueSalaryLoan = fields.Bool(required=True)
noPastDueOtherLoans = fields.Bool(required=True)
# RAC Check Schema
class RACCheckSchema(Schema):
transactionId = fields.Str(required=True)
fbnTransactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
RAC_Array = fields.List(fields.Nested(RACItemSchema), required=True)
app/schemas/repayment.py
+11
View File
@@ -0,0 +1,11 @@
from marshmallow import Schema, fields
# Repayment Schema
class RepaymentSchema(Schema):
type = fields.Str(data_key="$type", required=True)
msisdn = fields.Str(required=False) #optional
debtId = fields.Str(required=True)
productId = fields.Str(required=True)
transactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
channel = fields.Str(required=True)
app/schemas/revoke_enable_consent.py
+13
View File
@@ -0,0 +1,13 @@
from marshmallow import Schema, fields
# Revoke Enable Consent Schema
class RevokeEnableConsentSchema(Schema):
transactionId = fields.Str(required=True, metadata={"description": "Unique identifier of transaction in Simbrella system"})
fbnTransactionId = fields.Str(required=True, metadata={"description": "Unique id of the transaction received from FBN in CustomerConsentRequest"})
customerId = fields.Str(required=True, metadata={"description": "Unique identifier of a user"})
accountId = fields.Str(required=True, metadata={"description": "Specific identifier of a user’s account"})
processTime = fields.DateTime(required=True, metadata={"description": "Date and time when consent request was processed"})
consentType = fields.Str(required=True, metadata={"description": '“Enable” or “Revoke”'})
countryId = fields.Str(required=True, metadata={"description": 'Set to static value "01"'})
comment = fields.Str(required=False, metadata={"description": "Any additional comment for consent operation"})
app/schemas/select_offer.py
+13
View File
@@ -0,0 +1,13 @@
from marshmallow import Schema, fields
# Select Offer Schema
class SelectOfferSchema(Schema):
requestId = fields.Str(required=True, description="Unique request identifier")
transactionId = fields.Str(required=True, description="Transaction ID")
customerId = fields.Str(required=True, description="Customer ID")
accountId = fields.Str(required=True, description="Account ID")
msisdn = fields.Str(required=True, description="Mobile number")
requestedAmount = fields.Float(required=True, description="Amount requested")
productId = fields.Str(required=True, description="Product ID")
channel = fields.Str(required=True, description="Transaction channel (e.g., USSD)")
app/schemas/sms.py
+7
View File
@@ -0,0 +1,7 @@
from marshmallow import Schema, fields
# SMS Schema
class SMSSchema(Schema):
text = fields.Str(required=True)
dest = fields.Str(required=True)
unicode = fields.Bool(required=True)
app/schemas/token_validation.py
+8
View File
@@ -0,0 +1,8 @@
from marshmallow import Schema, fields
# Token Validation Schema
class TokenValidationSchema(Schema):
RequestId = fields.Str(required=True)
UserId = fields.Str(required=True)
CountryId = fields.Str(required=True)
TokenCode = fields.Str(required=True)
app/schemas/transaction_verify.py
+12
View File
@@ -0,0 +1,12 @@
from marshmallow import Schema, fields
# Transaction Verify Schema
class TransactionVerifySchema(Schema):
counter = fields.Str(required=True)
TransactionId = fields.Str(required=True)
requestID = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
countryId = fields.Str(required=True) # Static value “01”
transactionType = fields.Str(required=True)
app/static/css/main.css
View File
app/static/js/main.js
View File
app/templates/index.html
+11
View File
@@ -0,0 +1,11 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link rel="stylesheet" href="{{ url_for('static', filename='css/main.css') }}">
</head>
<body>
</body>
</html>
app/utils/__pycache__/logger.cpython-312.pyc
BIN
View File
Binary file not shown.
app/utils/logger.py
+13
View File
@@ -0,0 +1,13 @@
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(), # Log to console
logging.FileHandler("app.log", mode='a') # Log to file
]
)
logger = logging.getLogger("DetectionService")