DigiFi/digifi-BankToProductCore
1
1
Fork 1
Code Issues Pull Requests 2 Releases Activity

Merge branch 'bank-simbrella-calls' of DigiFi/digifi-BankToProductCore into master

Browse Source
This commit is contained in:
Olusesan Ameye
2025-04-01 10:18:19 +00:00
committed by Gogs
parent 41109d5353 39d1a1eddc
commit 4f2f3cccb4
33 changed files with 892 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__init__.py
+10 -2
View File
@@ -5,6 +5,9 @@ from flask_cors import CORS
from app.config import Config
from app.api.routes import api
from app.errors import register_error_handlers
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from app.extensions import db, migrate
def create_app():
""" Factory function to create a Flask app instance """
@@ -30,8 +33,13 @@ def create_app():
# Error Handlers
register_error_handlers(app)
from . import models
# Database and Migrations
db.init_app(app)
migrate.init_app(app, db)
return app
app/api/enums/__init__.py
+1
View File
@@ -0,0 +1 @@
from .transaction_type import TransactionType
app/api/enums/transaction_type.py
+10
View File
@@ -0,0 +1,10 @@
from enum import Enum
class TransactionType(str, Enum):
ELIGIBILITY_CHECK = "eligibility_check"
CUSTOMER_CONSENT = "customer_consent"
LOAN_STATUS = "loan_status"
NOTIFICATION_CALLBACK = "notification_callback"
PROVIDE_LOAN = "provide_loan"
REPAYMENT = "repayment"
SELECT_OFFER = "select_offer"
app/api/integrations/__init__.py
+1
View File
@@ -0,0 +1 @@
from .simbrella import SimbrellaClient
app/api/integrations/simbrella.py
+46
View File
@@ -0,0 +1,46 @@
import requests
from app.utils.logger import logger
from app.config import settings
class SimbrellaClient:
BASE_URL = settings.SIMBRELLA_BASE_URL
@staticmethod
def rac_check(customer_id, account_id, transaction_id):
"""
Calls the RACCheck endpoit
"""
url = f"{SimbrellaClient.BASE_URL}/RACCheck"
payload = {
"customerId": customer_id,
"accountId": account_id,
"transactionId": transaction_id,
"RAC_Array": [
{
"salaryAccount": True,
"bvn": "12345678901",
"crc": False,
"crms": True,
"accountStatus": "active",
"lien": False,
"noBouncedCheck": True,
"existingLoan": False,
"whitelist": True,
"noPastDueSalaryLoan": True,
"noPastDueOtherLoans": False
}
]
}
try:
response = requests.post(url, json=payload, timeout=10)
# Raise an error for non-200 responses
# response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as err:
logger.error(f"RACCheck API call failed: {str(err)}", exc_info=True)
return {"error": "RACCheck API error", "details": str(err)}
app/api/services/base_service.py
+55
View File
@@ -0,0 +1,55 @@
from app.models import Customer, Account, Transaction
from app.api.enums import TransactionType
from flask import jsonify
from marshmallow import ValidationError
import logging
logger = logging.getLogger(__name__)
class BaseService:
TRANSACTION_TYPE = None
@classmethod
def validate_data(cls, data, schema):
"""
Validate input data based on the provided schema.
"""
logger.info(f"Processing {cls.TRANSACTION_TYPE} request")
return schema.load(data)
@classmethod
def get_or_create_customer(cls, validated_data):
"""
Check if a customer exists; if not, create one.
"""
customer = Customer.query.filter_by(id=validated_data.get("customerId")).first()
if not customer:
customer = Customer.create_customer(
id=validated_data.get("customerId"),
msisdn=validated_data.get("msisdn"),
country_code=validated_data.get("countryCode"),
account_id=validated_data.get("accountId"),
)
return customer
@classmethod
def validate_account_ownership(cls, account_id, customer_id):
"""
Check if the provided account belongs to the customer.
"""
is_valid = Account.is_valid_account(account_id, customer_id)
return is_valid
@classmethod
def log_transaction(cls, validated_data):
"""
Create a new transaction.
"""
return Transaction.create_transaction(
id=validated_data.get("transactionId"),
account_id=validated_data.get("accountId"),
type=cls.TRANSACTION_TYPE,
channel=validated_data.get("channel"),
)
app/api/services/customer_consent.py
+36 -15
View File
@@ -1,10 +1,15 @@
from flask import request, jsonify
from app.api.services.base_service import BaseService
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.schemas.customer_consent import CustomerConsentSchema
from app.api.schemas.customer_consent import CustomerConsentSchema
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
class CustomerConsentService:
class CustomerConsentService(BaseService):
TRANSACTION_TYPE = TransactionType.CUSTOMER_CONSENT
@staticmethod
def process_request(data):
"""
@@ -17,34 +22,50 @@ class CustomerConsentService:
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
validated_data = CustomerConsentService.validate_data(data, CustomerConsentSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
if(CustomerConsentService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = CustomerConsentService.log_transaction(validated_data = validated_data)
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
# 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}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
}) , 500
app/api/services/eligibility_check.py
+46 -13
View File
@@ -1,9 +1,14 @@
from flask import session, jsonify
from app.utils.logger import logger
from app.api.services.base_service import BaseService
from app.api.schemas.eligibility_check import EligibilityCheckSchema
from marshmallow import ValidationError
from app.api.enums import TransactionType
from app.api.integrations import SimbrellaClient
class EligibilityCheckService(BaseService):
TRANSACTION_TYPE = TransactionType.ELIGIBILITY_CHECK
class EligibilityCheckService:
@staticmethod
def process_request(data):
"""
@@ -16,11 +21,37 @@ class EligibilityCheckService:
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
validated_data = EligibilityCheckService.validate_data(data, EligibilityCheckSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
customer = EligibilityCheckService.get_or_create_customer(validated_data = validated_data)
if (EligibilityCheckService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = EligibilityCheckService.log_transaction(validated_data = validated_data)
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
# Call RACCheck
response = SimbrellaClient.rac_check(
customer_id = customer_id,
account_id = account_id,
transaction_id = transaction.id,
)
if "error" in response or response.get("status") != 200:
return jsonify({"message": "RACCheck failed", "error": response.get("message", response)}), 400
offers = [
{
@@ -51,19 +82,21 @@ class EligibilityCheckService:
"accountId": "ACN8263457"
}
# 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}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
app/api/services/loan_status.py
+34 -13
View File
@@ -1,9 +1,14 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.schemas.loan_status import LoanStatusSchema
from app.api.schemas.loan_status import LoanStatusSchema
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
class LoanStatusService(BaseService):
TRANSACTION_TYPE = TransactionType.LOAN_STATUS
class LoanStatusService:
@staticmethod
def process_request(data):
"""
@@ -16,11 +21,23 @@ class LoanStatusService:
dict: A standardized response.
"""
try:
logger.info("Processing LoanStatus request")
validated_data = LoanStatusService.validate_data(data, LoanStatusSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
# Validate input data using the imported schema
schema = LoanStatusSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
if (LoanStatusService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = LoanStatusService.log_transaction(validated_data = validated_data)
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
loans = [
{
@@ -46,21 +63,25 @@ class LoanStatusService:
}
# 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}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
}) , 500
app/api/services/notification_callback.py
+15 -2
View File
@@ -1,9 +1,13 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
from app.utils.logger import logger
from app.api.schemas.notification_callback import NotificationCallbackSchema
class NotificationCallbackService:
class NotificationCallbackService(BaseService):
TRANSACTION_TYPE = TransactionType.NOTIFICATION_CALLBACK
@staticmethod
def process_request(data):
"""
@@ -37,10 +41,19 @@ class NotificationCallbackService:
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
app/api/services/provide_loan.py
+33 -12
View File
@@ -1,9 +1,14 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
from app.utils.logger import logger
from app.api.schemas.provide_loan import ProvideLoanSchema
class ProvideLoanService:
class ProvideLoanService(BaseService):
TRANSACTION_TYPE = TransactionType.PROVIDE_LOAN
@staticmethod
def process_request(data):
"""
@@ -16,13 +21,24 @@ class ProvideLoanService:
dict: A standardized response.
"""
try:
logger.info("Processing ProvideLoan request")
validated_data = ProvideLoanService.validate_data(data, ProvideLoanSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
# Validate input data using the imported schema
schema = ProvideLoanSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
if (ProvideLoanService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = ProvideLoanService.log_transaction(validated_data = validated_data)
# Business logic - providing a loan
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
response_data = {
"requestId": "202111170001371256908",
"transactionId": "Tr201712RK9232P115",
@@ -34,21 +50,26 @@ class ProvideLoanService:
}
# return ResponseHelper.success(
# data=response_data,
# message="Loan successfully provided"
# )
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
}) , 500
app/api/services/repayment.py
+31 -7
View File
@@ -1,9 +1,13 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.utils.logger import logger
from app.api.schemas.repayment import RepaymentSchema
from app.api.schemas.repayment import RepaymentSchema
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
class RepaymentService(BaseService):
TRANSACTION_TYPE = TransactionType.REPAYMENT
class RepaymentService:
@staticmethod
def process_request(data):
"""
@@ -16,11 +20,22 @@ class RepaymentService:
dict: A standardized response.
"""
try:
logger.info("Processing Repayment request")
validated_data = RepaymentService.validate_data(data, RepaymentSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
# Validate input data using the Repayment schema
schema = RepaymentSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
if (RepaymentService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = RepaymentService.log_transaction(validated_data = validated_data)
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
# Simulated processing logic
response_data = {
@@ -39,10 +54,19 @@ class RepaymentService:
return response_data
except ValidationError as err:
logger.error(f"Validation Error: {err.messages}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
app/api/services/select_offer.py
+31 -12
View File
@@ -1,9 +1,13 @@
from flask import request, jsonify
from marshmallow import ValidationError
from app.api.services.base_service import BaseService
from app.api.enums import TransactionType
from app.utils.logger import logger
from app.api.schemas.select_offer import SelectOfferSchema
class SelectOfferService:
class SelectOfferService(BaseService):
TRANSACTION_TYPE = TransactionType.SELECT_OFFER
@staticmethod
def process_request(data):
"""
@@ -16,12 +20,23 @@ class SelectOfferService:
dict: A standardized response.
"""
try:
logger.info("Processing SelectOffer request")
validated_data = SelectOfferService.validate_data(data, SelectOfferSchema())
account_id = validated_data.get('accountId')
customer_id = validated_data.get('customerId')
# Validate input data using the imported schema
schema = SelectOfferSchema()
validated_data = schema.load(data) # Raises ValidationError if invalid
if (SelectOfferService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
transaction = SelectOfferService.log_transaction(validated_data = validated_data)
if not transaction:
logger.error(f"Failed to log transaction")
return jsonify({
"message": "Failed to log transaction."
}), 400
else:
return jsonify({
"message": "Invalid Customer or Account"
}), 400
offers = [
{
"offerId": "14451",
@@ -54,21 +69,25 @@ class SelectOfferService:
}
# 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}")
logger.error(f"Validation Error: {getattr(err, 'messages', str(err))}")
return jsonify({
"message": "Validation exception"
}) , 422
except ValueError as err:
logger.error(f"{getattr(err, 'messages', str(err))}")
return jsonify({
"message": str(err)
}) , 400
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return jsonify({
"message": "Internal Server Error"
}) , 500
}) , 500
app/config.py
+16 -4
View File
@@ -3,9 +3,6 @@ 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")
SWAGGER_URL = os.getenv("SWAGGER_URL", "/documentation")
API_URL = os.getenv("API_URL", "/swagger.json")
@@ -14,4 +11,19 @@ class Config:
VALID_APP_ID = os.getenv("VALID_APP_ID", "app1")
VALID_API_KEY = os.getenv("VALID_API_KEY", "test-api-key-12345")
BASIC_AUTH_USERNAME = os.environ.get("BASIC_AUTH_USERNAME", "user")
BASIC_AUTH_PASSWORD = os.environ.get("BASIC_AUTH_PASSWORD", "password")
BASIC_AUTH_PASSWORD = os.environ.get("BASIC_AUTH_PASSWORD", "password")
DATABASE_USER = os.environ.get("DATABASE_USER")
DATABASE_PASSWORD = os.environ.get("DATABASE_PASSWORD")
DATABASE_HOST = os.environ.get("DATABASE_HOST")
DATABASE_PORT = os.environ.get("DATABASE_PORT", 10532)
DATABASE_NAME = os.environ.get("DATABASE_NAME")
SQLALCHEMY_DATABASE_URI = (
f"postgresql+psycopg2://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"
)
SQLALCHEMY_TRACK_MODIFICATIONS = False
SIMBRELLA_BASE_URL = os.getenv("SIMBRELLA_BASE_URL", "http://127.0.0.1:6337")
settings = Config()
app/extensions.py
+5
View File
@@ -0,0 +1,5 @@
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
db = SQLAlchemy()
migrate = Migrate()
app/models/__init__.py
+6
View File
@@ -0,0 +1,6 @@
from .customer import Customer
from .account import Account
from .loan import Loan
from .transaction import Transaction
__all__ = ['Customer', 'Account', 'Loan', 'Transaction']
app/models/account.py
+37
View File
@@ -0,0 +1,37 @@
from datetime import datetime, timezone
from app.extensions import db
class Account(db.Model):
__tablename__ = 'accounts'
id = db.Column(db.String(50), primary_key=True)
customer_id = db.Column(db.String(50), nullable=False)
account_type = db.Column(db.String(50))
status = db.Column(db.String(20), default='active')
lien_amount = db.Column(db.Float, default=0.0)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
@classmethod
def create_account(cls, id, customer_id, account_type, status='active'):
account = cls(
id=id,
customer_id=customer_id,
account_type=account_type
)
db.session.add(account)
db.session.commit()
return account
@classmethod
def is_valid_account(cls, account_id, customer_id):
account = cls.query.filter_by(id=account_id, customer_id=customer_id).first()
if not account:
return False
if account.lien_amount > 0:
return False
return True
def __repr__(self):
return f'<Account {self.id}>'
app/models/customer.py
+41
View File
@@ -0,0 +1,41 @@
from datetime import datetime, timezone
from app.extensions import db
from app.models.account import Account
class Customer(db.Model):
__tablename__ = 'customers'
id = db.Column(db.String(50), primary_key=True)
msisdn = db.Column(db.String(20), unique=True, nullable=False)
country_code = db.Column(db.String(3), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
@classmethod
def is_eligible(cls, customer_id):
customer = cls.query.filter_by(id=customer_id).first()
if not customer:
return False, "Customer not found"
return True, "Customer is eligible"
@classmethod
def create_customer(cls, id, msisdn, country_code, account_id, account_type='savings'):
if cls.query.filter_by(id=id).first():
raise ValueError("Customer already exists")
# Create the customer
customer = cls(id=id, msisdn=msisdn, country_code=country_code)
db.session.add(customer)
# Create an associated account
account = Account.create_account(
id=account_id,
customer_id=id,
account_type=account_type
)
db.session.commit()
return customer
def __repr__(self):
return f'<Customer {self.id}>'
app/models/loan.py
+31
View File
@@ -0,0 +1,31 @@
from datetime import datetime, timezone
from app.extensions import db
class Loan(db.Model):
__tablename__ = 'loans'
id = db.Column(db.String(50), primary_key=True)
customer_id = db.Column(db.String(50), nullable=False)
account_id = db.Column(db.String(50), nullable=False)
product_id = db.Column(db.String(20), nullable=False)
principal_amount = db.Column(db.Float, nullable=False)
status = db.Column(db.String(20), default='pending')
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
@classmethod
def has_active_loans(cls, customer_id):
active_loans = cls.query.filter_by(
customer_id=customer_id,
status='active'
).count()
if active_loans > 0:
return False, "Customer has active loans"
return True, "No active loans"
def __repr__(self):
return f'<Loan {self.id}>'
app/models/offer.py
+16
View File
@@ -0,0 +1,16 @@
from datetime import datetime, timezone
from app.extensions import db
class Offer(db.Model):
__tablename__ = 'offers'
id = db.Column(db.Integer, primary_key=True)
product_id = db.Column(db.String, nullable=False)
min_amount = db.Column(db.Float, nullable=False)
max_amount = db.Column(db.Float, nullable=False)
tenor = db.Column(db.Integer, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
def __repr__(self):
return f'<LoanOffer {self.id}>'
app/models/transaction.py
+41
View File
@@ -0,0 +1,41 @@
from datetime import datetime, timezone
from app.extensions import db
from sqlalchemy.exc import IntegrityError
class Transaction(db.Model):
__tablename__ = 'transactions'
id = db.Column(db.String(50), primary_key=True)
account_id = db.Column(db.String(50), nullable=False)
type = db.Column(db.String(50), nullable=False)
channel = db.Column(db.String(50), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
def __repr__(self):
return f'<Transaction {self.id}>'
@classmethod
def create_transaction(cls, id, account_id, type, channel):
if cls.query.filter_by(id=id).first():
raise ValueError("Duplicate Transaction")
transaction = cls(
id=id,
account_id=account_id,
type=type,
channel=channel
)
try:
db.session.add(transaction)
db.session.commit()
except IntegrityError as err:
db.session.rollback()
raise ValueError(f"Database integrity error: {err}")
return transaction
@classmethod
def get_transaction_by_id(cls, transaction_id):
return cls.query.get(transaction_id)