added extra query

This commit is contained in:
Chinenye Nmoh
2026-03-17 09:37:08 +01:00
parent bfb35e0285
commit 569d4c45d7
49 changed files with 4516 additions and 4505 deletions
+62 -62
View File
@@ -1,63 +1,63 @@
import random
import string
from app.services.repayment import RepaymentService
from app.services.loan import LoanService
from app.helpers.response_helper import ResponseHelper
from app.utils.logger import logger
from app.config import settings
OVERRIDE_COLLECTION_TRANCATION_ID = settings.OVERRIDE_COLLECTION_TRANCATION_ID
class CollectLoanHelper:
@staticmethod
def _validate_repayment_and_loan(data):
repayment = RepaymentService.get_repayment_by_id(id=data['Id'])
if not repayment:
logger.info(f"Repayment id: {data['Id']}, was not found")
return None, None, ResponseHelper.error("Repayment not found")
repayment_data = repayment.to_dict()
loan = LoanService.get_loan_by_loan_id(loan_id=int(repayment_data['loanId']))
if not loan:
logger.info(f"Loan id: {repayment_data['loanId']}, was not found")
return None, None, ResponseHelper.error("Loan not found")
loan
return repayment_data, loan, None
@staticmethod
def _build_collect_loan_payload(loan_data, repayment_data, data, collectionMethod):
logger.info(f"building CollectLoan endpoint with data: {loan_data}")
debtId = str(loan_data.get('debtId', "")).strip().zfill(6)
#this can be overridden based on config
t_id = ''.join(random.choices(string.ascii_uppercase, k=22))
if OVERRIDE_COLLECTION_TRANCATION_ID == 100:
t_id = loan_data['transactionId']
return {
"transactionId": t_id,
"fbnTransactionId": loan_data['transactionId'],
"debtId": debtId,
"customerId": repayment_data['customerId'],
"accountId": loan_data['accountId'],
"productId": repayment_data['productId'],
"collectAmount": (
data['overdueLoanScheduleAmount']
if data.get('overdueLoanScheduleAmount')
is not None else loan_data.get('balance', 0)
),
"penalCharge": 0,
"channel": "USSD",
"collectionMethod": collectionMethod,
"lienAmount": 0,
"countryId": "NG",
"comment": "COLLECT LOAN"
}
@staticmethod
def chunk_list(data, chunk_size):
"""Yield successive chunk_size chunks from data."""
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
import random
import string
from app.services.repayment import RepaymentService
from app.services.loan import LoanService
from app.helpers.response_helper import ResponseHelper
from app.utils.logger import logger
from app.config import settings
OVERRIDE_COLLECTION_TRANCATION_ID = settings.OVERRIDE_COLLECTION_TRANCATION_ID
class CollectLoanHelper:
@staticmethod
def _validate_repayment_and_loan(data):
repayment = RepaymentService.get_repayment_by_id(id=data['Id'])
if not repayment:
logger.info(f"Repayment id: {data['Id']}, was not found")
return None, None, ResponseHelper.error("Repayment not found")
repayment_data = repayment.to_dict()
loan = LoanService.get_loan_by_loan_id(loan_id=int(repayment_data['loanId']))
if not loan:
logger.info(f"Loan id: {repayment_data['loanId']}, was not found")
return None, None, ResponseHelper.error("Loan not found")
loan
return repayment_data, loan, None
@staticmethod
def _build_collect_loan_payload(loan_data, repayment_data, data, collectionMethod):
logger.info(f"building CollectLoan endpoint with data: {loan_data}")
debtId = str(loan_data.get('debtId', "")).strip().zfill(6)
#this can be overridden based on config
t_id = ''.join(random.choices(string.ascii_uppercase, k=22))
if OVERRIDE_COLLECTION_TRANCATION_ID == 100:
t_id = loan_data['transactionId']
return {
"transactionId": t_id,
"fbnTransactionId": loan_data['transactionId'],
"debtId": debtId,
"customerId": repayment_data['customerId'],
"accountId": loan_data['accountId'],
"productId": repayment_data['productId'],
"collectAmount": (
data['overdueLoanScheduleAmount']
if data.get('overdueLoanScheduleAmount')
is not None else loan_data.get('balance', 0)
),
"penalCharge": 0,
"channel": "USSD",
"collectionMethod": collectionMethod,
"lienAmount": 0,
"countryId": "NG",
"comment": "COLLECT LOAN"
}
@staticmethod
def chunk_list(data, chunk_size):
"""Yield successive chunk_size chunks from data."""
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
+250 -250
View File
@@ -1,251 +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.
"""
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)