DigiFi/digifi-BankEmulator
1
0
Fork 0
Code Issues Pull Requests Releases Activity

[add]: Swagger Documentation

Browse Source
This commit is contained in:
VivianDee
2025-03-24 08:44:57 +01:00
parent 7a4a9a9b99
commit dad8e68c9f
72 changed files with 1679 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/helpers/response_helper.py
+251
View File
@@ -0,0 +1,251 @@
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 jsonify(response), status_code
@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)
@staticmethod
def method_not_allowed(
message: str = "Method Not Allowed",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for a method not allowed 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, 405, error)
@staticmethod
def bad_request(
message: str = "Bad Request",
data: Optional[Union[Dict, List, str]] = None,
error: Optional[Union[Dict, str]] = None,
) -> Dict[str, Any]:
"""
Return a response for a bad request 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, 400, error)
app/api/middlewares/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .verify_api_key import require_api_key
from .app_id_checker import require_app_id
from .cors import enforce_json
app/api/middlewares/app_id_checker.py
+26
View File
@@ -0,0 +1,26 @@
from functools import wraps
from flask import request, jsonify
from app.utils.logger import logger
import os
# Load valid App-IDs from environment variables (comma-separated list)
VALID_APP_ID = os.getenv("VALID_APP_ID", "app1,app2,app3").split(",")
def require_app_id(f):
"""Decorator to enforce App-ID validation."""
@wraps(f)
def decorated_function(*args, **kwargs):
app_id = request.headers.get("App-ID")
if not app_id:
logger.error("Unauthorized access: Missing App-ID.")
return jsonify({"message": "Invalid request parameters"}), 400
if app_id not in VALID_APP_ID:
logger.error(f"Unauthorized access: Invalid App-ID {app_id}.")
return jsonify({"message": "Invalid request parameters"}), 400
return f(*args, **kwargs)
return decorated_function
app/api/middlewares/cors.py
+7
View File
@@ -0,0 +1,7 @@
from flask import request, jsonify
def enforce_json():
"""Middleware to enforce JSON Content-Type for incoming requests"""
if request.method in ["POST", "PUT", "PATCH"] and request.content_type != "application/json":
return jsonify({"message": "Invalid request parameters"}), 400
app/api/middlewares/verify_api_key.py
+26
View File
@@ -0,0 +1,26 @@
from functools import wraps
from flask import request, jsonify
from app.utils.logger import logger
import os
# Load valid API key from environment variables (fallback for testing)
VALID_API_KEY = os.getenv("VALID_API_KEY", "test-api-key-12345")
def require_api_key(f):
"""Decorator to enforce API key authentication."""
@wraps(f)
def decorated_function(*args, **kwargs):
api_key = request.headers.get("X-API-KEY")
if not api_key:
logger.error("Unauthorized access: Missing API key.")
return jsonify({"message": "Invalid request parameters"}), 400
if api_key != VALID_API_KEY:
logger.error("Unauthorized access: Invalid API key.")
return jsonify({"message": "Invalid request parameters"}), 400
return f(*args, **kwargs)
return decorated_function
app/api/routes/__init__.py
+1
View File
@@ -0,0 +1 @@
from .routes import api
app/api/routes/routes.py
+134
View File
@@ -0,0 +1,134 @@
from flask import Flask, Blueprint, request, jsonify, send_from_directory
import os
from app.api.services import (
RACCheckService,
DisbursementService,
CollectLoanService,
TransactionVerifyService,
PenalChargeService,
RevokeEnableConsentService,
TokenValidationService,
LienCheckService,
NewTransactionCheckService,
)
from app.utils.logger import logger
from app.api.middlewares import require_api_key, require_app_id, enforce_json
api = Blueprint("api", __name__)
# Enforce json
@api.before_request
def cors_middleware():
"""Middleware applied globally to all API routes in this blueprint"""
return enforce_json()
# Swagger JSON file
@api.route("/swagger.json", methods=['GET'])
def swagger_json():
swagger_dir = os.path.join("swagger")
return send_from_directory(swagger_dir, "digifi_swagger.json")
@api.route('/swagger/<path:filename>')
def serve_paths(filename):
swagger_dir = os.path.join("swagger")
return send_from_directory(swagger_dir, filename)
# RACCheck Endpoint
@api.route('/RACCheck', methods=['POST'])
@require_api_key
@require_app_id
def rac_check():
data = request.get_json()
# logger.info(f"RACCheck request received: {data}")
response = RACCheckService.process_request(data)
return response
# Disbursement Endpoint
@api.route('/Disbursement', methods=['POST'])
@require_api_key
@require_app_id
def disbursement():
data = request.get_json()
# logger.info(f"Disbursement request received: {data}")
response = DisbursementService.process_request(data)
return response
# CollectLoan Endpoint
@api.route('/CollectLoan', methods=['POST'])
@require_api_key
@require_app_id
def collect_loan():
data = request.get_json()
# logger.info(f"CollectLoan request received: {data}")
response = CollectLoanService.process_request(data)
return response
# TransactionVerify Endpoint
@api.route('/TransactionVerify', methods=['POST'])
@require_api_key
@require_app_id
def transaction_verify():
data = request.get_json()
# logger.info(f"TransactionVerify request received: {data}")
response = TransactionVerifyService.process_request(data)
return response
# PenalCharge Endpoint
@api.route('/PenalCharge', methods=['POST'])
@require_api_key
@require_app_id
def penal_charge():
data = request.get_json()
# logger.info(f"PenalCharge request received: {data}")
response = PenalChargeService.process_request(data)
return response
# RevokeEnableConsent Endpoint
@api.route('/RevokeEnableConsent', methods=['POST'])
@require_api_key
@require_app_id
def revoke_enable_consent():
data = request.get_json()
# logger.info(f"RevokeEnableConsent request received: {data}")
response = RevokeEnableConsentService.process_request(data)
return response
# TokenValidation Endpoint
@api.route('/TokenValidation', methods=['POST'])
@require_api_key
@require_app_id
def token_validation():
data = request.get_json()
# logger.info(f"TokenValidation request received: {data}")
response = TokenValidationService.process_request(data)
return response
# LienCheck Endpoint
@api.route('/LienCheck', methods=['POST'])
@require_api_key
@require_app_id
def lien_check():
data = request.get_json()
# logger.info(f"LienCheck request received: {data}")
response = LienCheckService.process_request(data)
return response
# NewTransactionCheck Endpoint
@api.route('/NewTransactionCheck', methods=['POST'])
@require_api_key
@require_app_id
def new_transaction_check():
data = request.get_json()
# logger.info(f"NewTransactionCheck request received: {data}")
response = NewTransactionCheckService.process_request(data)
return response
# Health Check Endpoint
@api.route('/health', methods=['GET'])
def health_check():
return {"status": "ok"} , 200
app/api/schemas/__init__.py
View File
app/api/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/api/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(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/api/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)
debtId = fields.Str(required=True)
transactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
productId = fields.Str(required=True)
provideAmount = fields.Float(required=True)
collectAmountInterest = fields.Float(required=False) # Optional
collectAmountMgtFee = fields.Float(required=True)
collectAmountInsurance = fields.Float(required=True)
collectAmountVAT = fields.Float(required=True)
countryId = fields.Str(required=True)
comment = fields.Str(required=False) # Optional
app/api/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)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
countryId = fields.Str(required=True)
app/api/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/api/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)
fbnTransactionId = fields.Str(required=True)
debtId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
penalCharge = fields.Decimal(required=True)
lienAmount = fields.Decimal(required=True)
countryId = fields.Str(required=True)
comment = fields.Str(required=False)
app/api/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/api/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)
fbnTransactionId = fields.Str(required=True)
customerId = fields.Str(required=True)
accountId = fields.Str(required=True)
processTime = fields.DateTime(required=True)
consentType = fields.Str(required=True)
countryId = fields.Str(required=True)
comment = fields.Str(required=False)
app/api/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/api/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/api/services/__init__.py
+9
View File
@@ -0,0 +1,9 @@
from app.api.services.rac_check import RACCheckService
from app.api.services.disbursement import DisbursementService
from app.api.services.collect_loan import CollectLoanService
from app.api.services.transaction_verify import TransactionVerifyService
from app.api.services.penal_charge import PenalChargeService
from app.api.services.revoke_enable_consent import RevokeEnableConsentService
from app.api.services.token_validation import TokenValidationService
from app.api.services.lien_check import LienCheckService
from app.api.services.new_transaction_check import NewTransactionCheckService
app/api/services/collect_loan.py
+61
View File
@@ -0,0 +1,61 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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)
# 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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/customer_consent.py
+51
View File
@@ -0,0 +1,51 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/disbursement.py
+62
View File
@@ -0,0 +1,62 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 = {
"transactionId": "T001",
"TransactionId": "Tr201712RK9232P115",
"debtId": "273194670",
"customerId": "CN621868",
"accountId": "2017821799",
"productId": "101",
"provideAmount": 100000.0,
"collectAmountInterest": 5000,
"collectAmountMgtFee": 1000,
"collectAmountInsurance": 1000,
"collectAmountVAT": 75,
"countryId": "01",
"resultCode": "00",
"resultDescription": "Loan Request Completed Successfully!"
}
# 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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/lien_check.py
+51
View File
@@ -0,0 +1,51 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.schemas.lien_check import LienCheckSchema
from flask import request, jsonify
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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/new_transaction_check.py
+58
View File
@@ -0,0 +1,58 @@
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.schemas.new_transaction_check import NewTransactionCheckSchema
from flask import request, jsonify
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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/penal_charge.py
+51
View File
@@ -0,0 +1,51 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/rac_check.py
+63
View File
@@ -0,0 +1,63 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/revoke_enable_consent.py
+54
View File
@@ -0,0 +1,54 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/token_validation.py
+54
View File
@@ -0,0 +1,54 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
app/api/services/transaction_verify.py
+57
View File
@@ -0,0 +1,57 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.helpers.response_helper import ResponseHelper
from app.api.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 jsonify({
"message": "Validation exception"
}) , 422
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500