Eamco/eamco_authorize
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor payment service, fix DB session, and consolidate endpoints

Browse Source 
- Fix critical NameError in database.py by restoring Session factory
- Refactor payment_service.py and crud.py to use shared constants.py and utils.py
- Deduplicate state mapping and input sanitization logic
- Move transaction amount calculation logic from CRUD to Router layer
- Enforce type safety in schemas using IntEnum for TransactionType/Status
- Move capture endpoint from transaction.py to payment.py (now /payments/capture)
- Update create_customer_profile signature for clarity
This commit is contained in:
Edwin Eames
2026-02-01 12:31:42 -05:00
parent 449eb74279
commit 97261f6c51
13 changed files with 335 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
app/constants.py Normal file
View File

@@ -0,0 +1,28 @@
"""
Constants shared across the EAMCO Authorize service.
"""
import enum
class TransactionStatus(enum.IntEnum):
"""Transaction status codes used throughout the payment system."""
APPROVED = 0
DECLINED = 1
class TransactionType(enum.IntEnum):
"""Transaction type codes for different payment operations."""
CHARGE = 0
AUTHORIZE = 1
CAPTURE = 2
# State ID to abbreviation mapping for Authorize.net billing address
# This maps the integer state IDs from the database to state abbreviations
STATE_ID_TO_ABBREVIATION = {
0: "MA",
1: "RI",
2: "NH",
3: "ME",
4: "VT",
5: "CT",
6: "NY"
}

39
app/crud.py
View File

@@ -1,17 +1,21 @@
## File: your_app/crud.py """
CRUD operations for the EAMCO Authorize service.
"""
import logging import logging
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from . import models, schemas from . import models, schemas
from .constants import TransactionStatus, TransactionType
from decimal import Decimal from decimal import Decimal
from typing import Optional, List
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# --- NEW CRUD FUNCTIONS FOR CIM --- # --- NEW CRUD FUNCTIONS FOR CIM ---
def get_card_by_id(db: Session, card_id: int): def get_card_by_id(db: Session, card_id: int) -> Optional[models.Card]:
return db.query(models.Card).filter(models.Card.id == card_id).first() return db.query(models.Card).filter(models.Card.id == card_id).first()
def update_customer_auth_net_profile_id(db: Session, customer_id: int, profile_id: str): def update_customer_auth_net_profile_id(db: Session, customer_id: int, profile_id: str) -> Optional[models.Customer]:
db_customer = get_customer(db, customer_id) db_customer = get_customer(db, customer_id)
if db_customer: if db_customer:
db_customer.auth_net_profile_id = profile_id db_customer.auth_net_profile_id = profile_id
@@ -19,13 +23,10 @@ def update_customer_auth_net_profile_id(db: Session, customer_id: int, profile_i
db.refresh(db_customer) db.refresh(db_customer)
return db_customer return db_customer
def create_customer_card(db: Session, customer_id: int, card_info: schemas.CardCreate, payment_profile_id: str): def create_customer_card(db: Session, customer_id: int, card_info: schemas.CardCreate, payment_profile_id: str) -> models.Card:
last_four_digits = card_info.card_number[-4:] last_four_digits = card_info.card_number[-4:]
try: # Schema guarantees YYYY-MM format
exp_year, exp_month = map(int, card_info.expiration_date.split('-')) exp_year, exp_month = map(int, card_info.expiration_date.split('-'))
except ValueError:
# Handle cases like "MM/YY" if necessary, but "YYYY-MM" is better
raise ValueError("Expiration date must be in YYYY-MM format")
db_card = models.Card( db_card = models.Card(
user_id=customer_id, user_id=customer_id,
@@ -57,10 +58,11 @@ def get_customer_by_email(db: Session, email: str):
def get_customers(db: Session, skip: int = 0, limit: int = 100): def get_customers(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Customer).offset(skip).limit(limit).all() return db.query(models.Customer).offset(skip).limit(limit).all()
def create_transaction(db: Session, transaction: schemas.TransactionBase, customer_id: int, status: int, auth_net_transaction_id: str = None): def create_transaction(db: Session, transaction: schemas.TransactionBase, customer_id: int, status: int, auth_net_transaction_id: str = None) -> models.Transaction:
# Using your existing logic for saving amounts # Logic for amounts moved to routers/payment.py
preauthorize_amount = transaction.preauthorize_amount if status == 0 else Decimal("0.0") # We simply save what is passed in the transaction object
charge_amount = transaction.charge_amount if (transaction.transaction_type != 1 and status == 0) else Decimal("0.0") preauthorize_amount = transaction.preauthorize_amount or Decimal("0.0")
charge_amount = transaction.charge_amount or Decimal("0.0")
db_transaction = models.Transaction( db_transaction = models.Transaction(
preauthorize_amount=preauthorize_amount, preauthorize_amount=preauthorize_amount,
@@ -71,6 +73,7 @@ def create_transaction(db: Session, transaction: schemas.TransactionBase, custom
auth_net_transaction_id=auth_net_transaction_id, auth_net_transaction_id=auth_net_transaction_id,
service_id=transaction.service_id, service_id=transaction.service_id,
delivery_id=transaction.delivery_id, delivery_id=transaction.delivery_id,
auto_id=transaction.auto_id,
card_id=transaction.card_id, card_id=transaction.card_id,
payment_gateway=transaction.payment_gateway, payment_gateway=transaction.payment_gateway,
rejection_reason=transaction.rejection_reason rejection_reason=transaction.rejection_reason
@@ -83,8 +86,8 @@ def create_transaction(db: Session, transaction: schemas.TransactionBase, custom
def get_transaction_by_delivery_id(db: Session, delivery_id: int): def get_transaction_by_delivery_id(db: Session, delivery_id: int):
return db.query(models.Transaction).filter( return db.query(models.Transaction).filter(
models.Transaction.delivery_id == delivery_id, models.Transaction.delivery_id == delivery_id,
models.Transaction.transaction_type == 1, models.Transaction.transaction_type == TransactionType.AUTHORIZE,
models.Transaction.status == 0 models.Transaction.status == TransactionStatus.APPROVED
).first() ).first()
def get_transaction_by_auth_id(db: Session, auth_net_transaction_id: str): def get_transaction_by_auth_id(db: Session, auth_net_transaction_id: str):
@@ -95,8 +98,8 @@ def get_transaction_by_auth_id(db: Session, auth_net_transaction_id: str):
def get_transaction_by_auto_id(db: Session, auto_id: int): def get_transaction_by_auto_id(db: Session, auto_id: int):
return db.query(models.Transaction).filter( return db.query(models.Transaction).filter(
models.Transaction.auto_id == auto_id, models.Transaction.auto_id == auto_id,
models.Transaction.transaction_type.in_([1, 2]), models.Transaction.transaction_type.in_([TransactionType.AUTHORIZE, TransactionType.CAPTURE]),
models.Transaction.status == 0 models.Transaction.status == TransactionStatus.APPROVED
).first() ).first()
def update_transaction_for_capture(db: Session, auth_net_transaction_id: str, charge_amount: Decimal, status: int, rejection_reason: str = None): def update_transaction_for_capture(db: Session, auth_net_transaction_id: str, charge_amount: Decimal, status: int, rejection_reason: str = None):
@@ -104,8 +107,8 @@ def update_transaction_for_capture(db: Session, auth_net_transaction_id: str, ch
if not transaction: if not transaction:
return None return None
transaction.charge_amount = charge_amount if status == 0 else Decimal("0.0") transaction.charge_amount = charge_amount if status == TransactionStatus.APPROVED else Decimal("0.0")
transaction.transaction_type = 2 transaction.transaction_type = TransactionType.CAPTURE
transaction.status = status transaction.status = status
if rejection_reason: if rejection_reason:
transaction.rejection_reason = rejection_reason transaction.rejection_reason = rejection_reason

7
app/database.py
View File

@@ -25,11 +25,10 @@ engine = create_engine(
pool_recycle=3600, # Recycle connections after 1 hour pool_recycle=3600, # Recycle connections after 1 hour
) )
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = Session()
Base = declarative_base() Base = declarative_base()
Base.metadata.create_all(engine)
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db(): def get_db():

4
app/models.py
View File

@@ -1,4 +1,4 @@
## File: your_app/models.py ## File: app/models.py
from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Numeric from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Numeric
from .database import Base from .database import Base
@@ -118,7 +118,7 @@ class Tickets_Auto_Delivery(Base):
price_per_gallon = Column(Numeric(10, 2), nullable=True) price_per_gallon = Column(Numeric(10, 2), nullable=True)
total_amount_customer = Column(Numeric(10, 2), nullable=True) total_amount_customer = Column(Numeric(10, 2), nullable=True)
customer_town = Column(String(140)) # customer_town removed (duplicate)
payment_type = Column(Integer, nullable=True) payment_type = Column(Integer, nullable=True)
payment_card_id = Column(Integer, nullable=True) payment_card_id = Column(Integer, nullable=True)

180
app/routers/payment.py
View File

@@ -1,133 +1,25 @@
## File: your_app/views.py """
Payment Router - Main payment processing endpoints for card management and transactions.
"""
import logging import logging
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
logger = logging.getLogger(__name__)
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from typing import Tuple, Optional
import enum
from decimal import Decimal
from .. import crud, models, schemas, database from .. import crud, models, schemas, database
from ..services import payment_service from ..services import payment_service
from ..services.payment_service import parse_authnet_response
from ..constants import TransactionStatus, TransactionType, STATE_ID_TO_ABBREVIATION
from config import load_config from config import load_config
logger = logging.getLogger(__name__)
ApplicationConfig = load_config() ApplicationConfig = load_config()
AuthNetResponse = object
router = APIRouter( router = APIRouter(
prefix="/payments", prefix="/payments",
tags=["Payments & Transactions"], tags=["Payments & Transactions"],
) )
class TransactionStatus(enum.IntEnum):
APPROVED = 0
DECLINED = 1
class TransactionType(enum.IntEnum):
CHARGE = 0
AUTHORIZE = 1
CAPTURE = 3
# --- NEW CIM CORE FUNCTIONS ---
STATE_ID_TO_ABBREVIATION = {
0: "MA",
1: "RI",
2: "NH",
3: "ME",
4: "VT",
5: "CT",
6: "NY"
}
def _parse_authnet_response(response: Optional[AuthNetResponse]) -> Tuple[TransactionStatus, Optional[str], Optional[str]]:
"""
Parse Authorize.net response with proper attribute access for SDK objects.
Authorize.net response objects don't have .text properties, they're direct attributes.
"""
logger.debug(f"DEBUG: Parsing response, type: {type(response)}")
logger.debug(f"DEBUG: Response exists: {response is not None}")
if response is not None:
logger.debug("DEBUG: Checking for messages attribute...")
if hasattr(response, 'messages'):
logger.debug(f"DEBUG: Messages exist, resultCode: {getattr(response.messages, 'resultCode', 'NO resultCode')}")
else:
logger.debug("DEBUG: No messages attribute")
if response.messages.resultCode == "Ok":
logger.debug("✅✅ DEBUG: Taking APPROVED path")
status = TransactionStatus.APPROVED
auth_net_transaction_id = None
# Extract transaction ID with proper error handling
try:
if hasattr(response, 'transactionResponse') and response.transactionResponse is not None:
if hasattr(response.transactionResponse, 'transId') and response.transactionResponse.transId:
auth_net_transaction_id = str(response.transactionResponse.transId)
logger.debug(f"✅✅ DEBUG: FOUND transaction ID: {auth_net_transaction_id}")
else:
logger.debug("DEBUG: transactionResponse exists but no transId")
else:
logger.debug("DEBUG: No transactionResponse in approved response")
except Exception as e:
logger.debug(f"DEBUG: Exception extracting transaction ID: {e}")
logger.debug(f"DEBUG: Response object inspection:")
logger.debug(type(response))
if hasattr(response, 'transactionResponse'):
logger.debug(f"TransactionResponse type: {type(response.transactionResponse)}")
logger.debug(dir(response.transactionResponse))
rejection_reason = None
logger.debug(f"✅✅✅ DEBUG: APPROVED - ID: {auth_net_transaction_id}, rejection: {rejection_reason}")
else:
logger.debug("DEBUG: Taking DECLINED path")
status = TransactionStatus.DECLINED
auth_net_transaction_id = None
rejection_reason = "Payment declined by gateway."
if response is not None:
# Handle transaction response errors
if hasattr(response, 'transactionResponse') and response.transactionResponse is not None:
if hasattr(response.transactionResponse, 'errors') and response.transactionResponse.errors:
logger.debug("DEBUG: Using transactionResponse.errors")
try:
error = response.transactionResponse.errors[0]
# Remove the .text access - use direct attributes
error_code = getattr(error, 'errorCode', 'Unknown')
error_text = getattr(error, 'errorText', 'Unknown error')
rejection_reason = f"{error_code}: {error_text}"
logger.debug(f"DEBUG: Transaction error: {rejection_reason}")
except Exception as e:
logger.debug(f"DEBUG: Exception parsing transaction error: {e}")
rejection_reason = "Failed to parse transaction error"
# Handle message-level errors
elif hasattr(response, 'messages') and response.messages:
if hasattr(response.messages, 'message') and response.messages.message:
logger.debug("DEBUG: Using response.messages.message")
try:
msg = response.messages.message
if isinstance(msg, list):
msg = msg[0] if msg else None
if msg:
code = getattr(msg, 'code', 'Unknown')
text = getattr(msg, 'text', 'Unknown error')
rejection_reason = f"{code}: {text}"
logger.debug(f"DEBUG: Message error: {rejection_reason}")
except Exception as e:
logger.debug(f"DEBUG: Exception parsing message error: {e}")
rejection_reason = "Failed to parse message error"
logger.debug(f"✅✅✅ DEBUG: FINAL RESULT - Status: {status}, ID: {auth_net_transaction_id}, Reason: {rejection_reason}")
return status, auth_net_transaction_id, rejection_reason
@router.post("/customers/{customer_id}/cards", summary="Add a new payment card for a customer") @router.post("/customers/{customer_id}/cards", summary="Add a new payment card for a customer")
def add_card_to_customer(customer_id: int, card_info: schemas.CardCreate, db: Session = Depends(database.get_db)): def add_card_to_customer(customer_id: int, card_info: schemas.CardCreate, db: Session = Depends(database.get_db)):
@@ -149,11 +41,19 @@ def add_card_to_customer(customer_id: int, card_info: schemas.CardCreate, db: Se
try: try:
# This part now works because the service hard-codes the state to "MA" # This part now works because the service hard-codes the state to "MA"
if not db_customer.auth_net_profile_id: if not db_customer.auth_net_profile_id:
profile_id, payment_id = payment_service.create_customer_profile( # 1. Create the customer profile (returns ID, but no payment profile yet)
profile_id = payment_service.create_customer_profile(
customer=customer_schema, card_info=card_info customer=customer_schema, card_info=card_info
) )
# 2. Update local DB with the new profile ID
crud.update_customer_auth_net_profile_id(db, customer_id=customer_id, profile_id=profile_id) crud.update_customer_auth_net_profile_id(db, customer_id=customer_id, profile_id=profile_id)
payment_profile_id = payment_id
# 3. Explicitly add the payment profile to get the payment_profile_id
payment_profile_id = payment_service.add_payment_profile_to_customer(
customer_profile_id=profile_id,
customer=customer_schema,
card_info=card_info
)
else: else:
payment_profile_id = payment_service.add_payment_profile_to_customer( payment_profile_id = payment_service.add_payment_profile_to_customer(
customer_profile_id=db_customer.auth_net_profile_id, customer_profile_id=db_customer.auth_net_profile_id,
@@ -262,10 +162,16 @@ def charge_saved_card(customer_id: int, transaction_req: schemas.TransactionCrea
transaction_req=transaction_req transaction_req=transaction_req
) )
status, auth_net_transaction_id, rejection_reason = _parse_authnet_response(auth_net_response)
status, auth_net_transaction_id, rejection_reason = parse_authnet_response(auth_net_response)
# Calculate amounts to save based on status (Business Logic moved from CRUD)
final_charge_amount = transaction_req.charge_amount if status == TransactionStatus.APPROVED else Decimal("0.0")
final_preauth_amount = Decimal("0.0")
transaction_data = schemas.TransactionBase( transaction_data = schemas.TransactionBase(
charge_amount=transaction_req.charge_amount, charge_amount=final_charge_amount,
preauthorize_amount=final_preauth_amount,
transaction_type=TransactionType.CHARGE, transaction_type=TransactionType.CHARGE,
service_id=transaction_req.service_id, service_id=transaction_req.service_id,
delivery_id=transaction_req.delivery_id, delivery_id=transaction_req.delivery_id,
@@ -355,8 +261,7 @@ def authorize_saved_card(customer_id: int, transaction_req: schemas.TransactionA
) )
# Check if the test authorization worked # Check if the test authorization worked
from ..services import payment_service as ps # Need access to _parse_authnet_response test_status, _, test_reason = parse_authnet_response(test_response)
test_status, _, test_reason = _parse_authnet_response(test_response)
if "E00121" in str(test_reason) or test_status == 1: # 1 = DECLINED if "E00121" in str(test_reason) or test_status == 1: # 1 = DECLINED
logger.debug(f"🐛 DEBUG: TEST AUTH FAILED - Payment profile exists but is INVALID!") logger.debug(f"🐛 DEBUG: TEST AUTH FAILED - Payment profile exists but is INVALID!")
@@ -427,11 +332,17 @@ def authorize_saved_card(customer_id: int, transaction_req: schemas.TransactionA
) )
# Parse the transaction response (no need for E00121 nuclear cleanup since pre-validation should have caught it) # Parse the transaction response (no need for E00121 nuclear cleanup since pre-validation should have caught it)
transaction_status, auth_net_transaction_id, rejection_reason = _parse_authnet_response(auth_net_response) transaction_status, auth_net_transaction_id, rejection_reason = parse_authnet_response(auth_net_response)
logger.debug(transaction_req) logger.debug(transaction_req)
# Calculate amounts to save based on status (Business Logic moved from CRUD)
final_preauth_amount = transaction_req.preauthorize_amount if transaction_status == TransactionStatus.APPROVED else Decimal("0.0")
final_charge_amount = Decimal("0.0")
transaction_data = schemas.TransactionBase( transaction_data = schemas.TransactionBase(
preauthorize_amount=transaction_req.preauthorize_amount, preauthorize_amount=final_preauth_amount,
charge_amount=final_charge_amount,
transaction_type=TransactionType.AUTHORIZE, # This is key transaction_type=TransactionType.AUTHORIZE, # This is key
service_id=transaction_req.service_id, service_id=transaction_req.service_id,
delivery_id=transaction_req.delivery_id, delivery_id=transaction_req.delivery_id,
@@ -449,3 +360,26 @@ def authorize_saved_card(customer_id: int, transaction_req: schemas.TransactionA
) )
return db_transaction return db_transaction
@router.post("/capture", response_model=schemas.Transaction, summary="Capture a previously authorized amount")
def capture_authorized_amount(transaction: schemas.TransactionCapture, db: Session = Depends(database.get_db)):
# This endpoint captures a previously authorized transaction
# It finds the original transaction by its ID and captures the funds
logger.info(f"POST /payments/capture - Capturing authorized transaction {transaction.auth_net_transaction_id}")
auth_transaction = crud.get_transaction_by_auth_id(db, auth_net_transaction_id=transaction.auth_net_transaction_id)
if not auth_transaction:
raise HTTPException(status_code=404, detail="Authorization transaction not found")
# Call the capture service function
auth_net_response = payment_service.capture_authorized_transaction(transaction)
status, _, rejection_reason = parse_authnet_response(auth_net_response)
# Use the existing CRUD function to update the transaction
return crud.update_transaction_for_capture(
db=db,
auth_net_transaction_id=transaction.auth_net_transaction_id,
charge_amount=transaction.charge_amount,
status=status,
rejection_reason=rejection_reason
)

50
app/routers/transaction.py
View File

@@ -1,26 +1,19 @@
## File: transaction.py (New transaction router) """
Transaction Router - Endpoints for transaction lookup and capture operations.
"""
import logging import logging
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
import enum
# Import locally to avoid circular imports
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
from app import crud, database, schemas, models from app import crud, database, schemas, models
from app.services import payment_service from app.services import payment_service
from app.services.payment_service import parse_authnet_response, TransactionStatus
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Create a router for transaction endpoints # Create a router for transaction endpoints
transaction_router = APIRouter() transaction_router = APIRouter()
class TransactionStatus(enum.IntEnum):
APPROVED = 0
DECLINED = 1
# Test endpoint to verify router is working # Test endpoint to verify router is working
@transaction_router.get("/test/", summary="Test transaction router") @transaction_router.get("/test/", summary="Test transaction router")
def test_transaction_router(): def test_transaction_router():
@@ -29,9 +22,6 @@ def test_transaction_router():
return {"test": "transaction router is working"} return {"test": "transaction router is working"}
@transaction_router.get("/transaction/delivery/{delivery_id}", summary="Get pre-authorization transaction for a delivery") @transaction_router.get("/transaction/delivery/{delivery_id}", summary="Get pre-authorization transaction for a delivery")
def get_delivery_transaction(delivery_id: int, db: Session = Depends(database.get_db)): def get_delivery_transaction(delivery_id: int, db: Session = Depends(database.get_db)):
""" """
@@ -84,36 +74,4 @@ def update_transaction_auto_id(transaction_id: int, new_auto_id: int, db: Sessio
return {"message": "Transaction auto_id updated"} return {"message": "Transaction auto_id updated"}
@transaction_router.post("/capture/", response_model=schemas.Transaction, summary="Capture a previously authorized amount")
def capture_authorized_amount(transaction: schemas.TransactionCapture, db: Session = Depends(database.get_db)):
# This endpoint captures a previously authorized transaction
# It finds the original transaction by its ID and captures the funds
logger.info(f"POST /capture - Capturing authorized transaction {transaction.auth_net_transaction_id}")
auth_transaction = crud.get_transaction_by_auth_id(db, auth_net_transaction_id=transaction.auth_net_transaction_id)
if not auth_transaction:
raise HTTPException(status_code=404, detail="Authorization transaction not found")
# Call the capture service function
auth_net_response = payment_service.capture_authorized_transaction(transaction)
status, _, rejection_reason = _parse_authnet_response(auth_net_response)
# Use the existing CRUD function to update the transaction
return crud.update_transaction_for_capture(
db=db,
auth_net_transaction_id=transaction.auth_net_transaction_id,
charge_amount=transaction.charge_amount,
status=status,
rejection_reason=rejection_reason
)
def _parse_authnet_response(response):
"""
Parse Authorize.Net response for transaction status
"""
if response.messages.resultCode == "Ok":
status = TransactionStatus.APPROVED
rejection_reason = None
else:
status = TransactionStatus.DECLINED
rejection_reason = "Payment declined by gateway."
return status, None, rejection_reason

8
app/schemas.py
View File

@@ -4,7 +4,9 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator
from typing import List, Optional from typing import List, Optional
from datetime import datetime from datetime import datetime
from decimal import Decimal from decimal import Decimal
from decimal import Decimal
import re import re
from .constants import TransactionType, TransactionStatus
# --- NEW SCHEMAS FOR CIM WORKFLOW (Now with correct Pydantic V2 config) --- # --- NEW SCHEMAS FOR CIM WORKFLOW (Now with correct Pydantic V2 config) ---
@@ -68,7 +70,7 @@ class TransactionAuthorizeByCardID(BaseModel):
class TransactionBase(BaseModel): class TransactionBase(BaseModel):
preauthorize_amount: Optional[Decimal] = None preauthorize_amount: Optional[Decimal] = None
charge_amount: Optional[Decimal] = None charge_amount: Optional[Decimal] = None
transaction_type: int transaction_type: TransactionType
service_id: Optional[int] = None service_id: Optional[int] = None
delivery_id: Optional[int] = None delivery_id: Optional[int] = None
auto_id: Optional[int] = None auto_id: Optional[int] = None
@@ -94,8 +96,8 @@ class TransactionCapture(BaseModel):
class Transaction(TransactionBase): class Transaction(TransactionBase):
id: int id: int
transaction_type: int transaction_type: TransactionType
status: int status: TransactionStatus
auth_net_transaction_id: Optional[str] = None auth_net_transaction_id: Optional[str] = None
customer_id: int customer_id: int
created_at: datetime created_at: datetime

34
app/services/check_user_service.py
View File

@@ -10,31 +10,10 @@ from sqlalchemy.orm import Session
# Set Authorize.net environment based on configuration # Set Authorize.net environment based on configuration
from authorizenet.constants import constants from config import load_config, API_LOGIN_ID, TRANSACTION_KEY, VALIDATION_MODE, ENVIRONMENT
# Set environment
from config import load_config # Assuming you have this constants.environment = ENVIRONMENT
# Load Authorize.net credentials
ApplicationConfig = load_config()
# Set Authorize.net environment based on configuration
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
else:
constants.environment = constants.SANDBOX
constants.show_url_on_request = True
VALIDATION_MODE = "testMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
@@ -193,13 +172,6 @@ def _get_customer_profile(profile_id: str):
controller = controllers.getCustomerProfileController(request) controller = controllers.getCustomerProfileController(request)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
response = controller.getresponse() response = controller.getresponse()

215
app/services/payment_service.py
View File

@@ -1,8 +1,15 @@
## File: your_app/services/payment_service.py """
Payment Service - Core Authorize.net API interactions for EAMCO.
This module provides the canonical implementations of payment processing functions
including transaction parsing, customer profile management, and payment operations.
"""
import logging import logging
import traceback import traceback
import re import re
import enum
from typing import Tuple, Optional
from authorizenet import apicontractsv1 from authorizenet import apicontractsv1
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -14,29 +21,19 @@ from authorizenet.apicontrollers import (
) )
from authorizenet.constants import constants from authorizenet.constants import constants
from .. import schemas from .. import schemas
from config import load_config # Assuming you have this from ..constants import TransactionStatus, TransactionType, STATE_ID_TO_ABBREVIATION
from ..utils import sanitize_input
from config import load_config
# Load Authorize.net credentials # Load Authorize.net credentials
from config import load_config, API_LOGIN_ID, TRANSACTION_KEY, VALIDATION_MODE, ENVIRONMENT
# Load Authorize.net environment variables
ApplicationConfig = load_config() ApplicationConfig = load_config()
# Set Authorize.net environment based on configuration # Set environment
# Set Authorize.net environment based on configuration constants.environment = ENVIRONMENT
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
else:
constants.environment = constants.SANDBOX
constants.show_url_on_request = True
VALIDATION_MODE = "testMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
@@ -98,16 +95,99 @@ def _get_authnet_error_message(response):
return "An unknown error occurred with the payment gateway." return "An unknown error occurred with the payment gateway."
def parse_authnet_response(response) -> Tuple[TransactionStatus, Optional[str], Optional[str]]:
"""
Parse Authorize.net response with proper attribute access for SDK objects.
This is the CANONICAL implementation - all routers should import this function
rather than defining their own version.
Args:
response: Authorize.net API response object
Returns:
Tuple of (status, auth_net_transaction_id, rejection_reason)
- status: TransactionStatus.APPROVED (0) or TransactionStatus.DECLINED (1)
- auth_net_transaction_id: Transaction ID string if successful, None otherwise
- rejection_reason: Error message string if declined, None if approved
"""
logger.debug(f"Parsing Authorize.net response, type: {type(response)}")
if response is None:
logger.error("Authorize.net response is None")
return TransactionStatus.DECLINED, None, "No response from payment gateway"
if not hasattr(response, 'messages') or response.messages is None:
logger.error("Authorize.net response missing messages attribute")
return TransactionStatus.DECLINED, None, "Invalid response from payment gateway"
if response.messages.resultCode == "Ok":
logger.debug("Response resultCode is Ok - APPROVED path")
status = TransactionStatus.APPROVED
auth_net_transaction_id = None
rejection_reason = None
# Extract transaction ID with proper error handling
try:
if hasattr(response, 'transactionResponse') and response.transactionResponse is not None:
if hasattr(response.transactionResponse, 'transId') and response.transactionResponse.transId:
auth_net_transaction_id = str(response.transactionResponse.transId)
logger.debug(f"Extracted transaction ID: {auth_net_transaction_id}")
else:
logger.debug("transactionResponse exists but no transId found")
else:
logger.debug("No transactionResponse in approved response")
except Exception as e:
logger.warning(f"Exception extracting transaction ID: {e}")
return status, auth_net_transaction_id, rejection_reason
else:
logger.debug("Response resultCode is not Ok - DECLINED path")
status = TransactionStatus.DECLINED
auth_net_transaction_id = None
rejection_reason = "Payment declined by gateway."
# Handle transaction response errors (most specific)
if hasattr(response, 'transactionResponse') and response.transactionResponse is not None:
if hasattr(response.transactionResponse, 'errors') and response.transactionResponse.errors:
try:
error = response.transactionResponse.errors[0]
error_code = getattr(error, 'errorCode', 'Unknown')
error_text = getattr(error, 'errorText', 'Unknown error')
rejection_reason = f"{error_code}: {error_text}"
logger.debug(f"Transaction error: {rejection_reason}")
except Exception as e:
logger.warning(f"Exception parsing transaction error: {e}")
# Handle message-level errors (fallback)
elif hasattr(response, 'messages') and response.messages:
if hasattr(response.messages, 'message') and response.messages.message:
try:
msg = response.messages.message
if isinstance(msg, list):
msg = msg[0] if msg else None
if msg:
code = getattr(msg, 'code', 'Unknown')
text = getattr(msg, 'text', 'Unknown error')
rejection_reason = f"{code}: {text}"
logger.debug(f"Message error: {rejection_reason}")
except Exception as e:
logger.warning(f"Exception parsing message error: {e}")
return status, auth_net_transaction_id, rejection_reason
def create_customer_profile(customer: schemas.Customer, card_info: schemas.CardCreate): def create_customer_profile(customer: schemas.Customer, card_info: schemas.CardCreate):
""" """
Creates a new customer profile in Authorize.Net (payment profiles added separately). Creates a new customer profile in Authorize.Net (payment profiles added separately).
This version sanitizes and trims customer data before sending. This version sanitizes and trims customer data before sending.
""" """
# Note: Never log API credentials
try: try:
merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY)
except Exception as e: except Exception as e:
pass # Will be handled by request failure below logger.error(f"Failed to create merchant authentication: {e}")
raise ValueError("Payment gateway authentication failed. Please check API credentials.")
# API max lengths: email=255 # API max lengths: email=255
email = (customer.customer_email or f"no-email-{customer.id}@example.com")[:255] email = (customer.customer_email or f"no-email-{customer.id}@example.com")[:255]
@@ -126,19 +206,12 @@ def create_customer_profile(customer: schemas.Customer, card_info: schemas.CardC
controller = createCustomerProfileController(request) controller = createCustomerProfileController(request)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
response = controller.getresponse() response = controller.getresponse()
# Check if response is None (API call failed) # Check if response is None (API call failed)
if response is None: if response is None:
logger.debug("ERROR: Authorize.net API call returned None - likely a network/connectivity issue") logger.error("ERROR: Authorize.net API call returned None - likely a network/connectivity issue")
raise ValueError("Could not connect to the payment gateway. Please check network connectivity.") raise ValueError("Could not connect to the payment gateway. Please check network connectivity.")
try: try:
@@ -151,9 +224,9 @@ def create_customer_profile(customer: schemas.Customer, card_info: schemas.CardC
# Add detailed logging # Add detailed logging
logger.debug(f"API Response - Profile ID: {profile_id}") logger.debug(f"API Response - Profile ID: {profile_id}")
logger.debug(f"Returning: profile_id='{str(profile_id)}', payment_id=''") logger.debug(f"Returning: profile_id='{str(profile_id)}'")
return str(profile_id), "" return str(profile_id)
else: else:
error_msg = _get_authnet_error_message(response) error_msg = _get_authnet_error_message(response)
logger.debug(f"Failed to create customer profile (API Error): {error_msg}") logger.debug(f"Failed to create customer profile (API Error): {error_msg}")
@@ -190,7 +263,6 @@ def authorize_customer_profile(customer_profile_id: str, payment_profile_id: str
# CHECK FOR E00121 ERROR - "invalid payment profile ID" # CHECK FOR E00121 ERROR - "invalid payment profile ID"
if db_session and customer_id and card_id and _is_e00121_response(response): if db_session and customer_id and card_id and _is_e00121_response(response):
logger.debug(f"E00121 DETECTED! Invalid payment profile {payment_profile_id}") logger.debug(f"E00121 DETECTED! Invalid payment profile {payment_profile_id}")
logger.debug("POOOP")
logger.debug(f"AUTO-RECOVERING: Starting payment profile refresh for customer {customer_id}") logger.debug(f"AUTO-RECOVERING: Starting payment profile refresh for customer {customer_id}")
try: try:
@@ -256,13 +328,6 @@ def _perform_authorization(customer_profile_id: str, payment_profile_id: str, tr
controller = createTransactionController(createtransactionrequest) controller = createTransactionController(createtransactionrequest)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
response = controller.getresponse() response = controller.getresponse()
@@ -298,13 +363,6 @@ def capture_authorized_transaction(transaction_req: schemas.TransactionCapture):
) )
controller = createTransactionController(createtransactionrequest) controller = createTransactionController(createtransactionrequest)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
return controller.getresponse() return controller.getresponse()
@@ -315,39 +373,21 @@ def add_payment_profile_to_customer(customer_profile_id: str, customer: schemas.
merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY)
def sanitize(text, max_len, allow_spaces=False, is_zip=False): first_name = sanitize_input(customer.customer_first_name, 50) or "N/A"
if not text: last_name = sanitize_input(customer.customer_last_name, 50) or "N/A"
return "" address = sanitize_input(customer.customer_address, 60, allow_spaces=True) or "N/A"
if is_zip: city = sanitize_input(customer.customer_town, 40) or "N/A"
pattern = r'[^a-zA-Z0-9-]'
else:
pattern = r'[^a-zA-Z0-9]' if not allow_spaces else r'[^a-zA-Z0-9\s]'
sanitized = re.sub(pattern, '', str(text))
return sanitized.strip()[:max_len]
first_name = sanitize(customer.customer_first_name, 50) or "N/A" # Convert state ID (integer) to abbreviation using the mapping
last_name = sanitize(customer.customer_last_name, 50) or "N/A" # customer.customer_state is an integer foreign key, not a string
address = sanitize(customer.customer_address, 60, allow_spaces=True) or "N/A" state = STATE_ID_TO_ABBREVIATION.get(customer.customer_state, "MA")
city = sanitize(customer.customer_town, 40) or "N/A"
# ========= CHANGE 1.B: ADD STATE HERE ========= zip_code = sanitize_input(customer.customer_zip, 20, is_zip=True)
state = sanitize(customer.customer_state, 40) or "MA" # Defaulting to MA for safety
zip_code = sanitize(customer.customer_zip, 20, is_zip=True) # Parse expiration date - expected format is "YYYY-MM" (validated by schema)
# Fix expiration date format for cards
try:
expiration_year = int(card_info.expiration_date.split('-')[0]) expiration_year = int(card_info.expiration_date.split('-')[0])
expiration_month = int(card_info.expiration_date.split('-')[1]) expiration_month = int(card_info.expiration_date.split('-')[1])
expiration_date = f"{expiration_month:02d}{expiration_year % 100:02d}" expiration_date = f"{expiration_month:02d}{expiration_year % 100:02d}"
except (ValueError, IndexError):
sanitized_exp = card_info.expiration_date.replace('/', '').replace('-', '')
if len(sanitized_exp) == 4:
expiration_date = sanitized_exp
else:
expiration_date = "0325"
# Expiration date parsed successfully
creditCard = apicontractsv1.creditCardType( creditCard = apicontractsv1.creditCardType(
cardNumber=card_info.card_number, cardNumber=card_info.card_number,
@@ -382,20 +422,12 @@ def add_payment_profile_to_customer(customer_profile_id: str, customer: schemas.
controller = createCustomerPaymentProfileController(request) controller = createCustomerPaymentProfileController(request)
try: try:
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute() controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute()
response = controller.getresponse() response = controller.getresponse()
# Check if response is None (API call failed) # Check if response is None (API call failed)
if response is None: if response is None:
logger.debug("ERROR: Authorize.net API call returned None - likely a network/connectivity issue") logger.error("ERROR: Authorize.net API call returned None - likely a network/connectivity issue")
raise ValueError("Could not connect to the payment gateway. Please check network connectivity.") raise ValueError("Could not connect to the payment gateway. Please check network connectivity.")
if response.messages.resultCode == "Ok": if response.messages.resultCode == "Ok":
@@ -433,20 +465,12 @@ def get_customer_payment_profiles(customer_profile_id: str):
controller = getCustomerProfileController(request) controller = getCustomerProfileController(request)
try: try:
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute() controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute()
response = controller.getresponse() response = controller.getresponse()
# Check if response is None (API call failed) # Check if response is None (API call failed)
if response is None: if response is None:
logger.debug("ERROR: Authorize.net API call returned None - likely a network/connectivity issue") logger.error("ERROR: Authorize.net API call returned None - likely a network/connectivity issue")
raise ValueError("Could not connect to the payment gateway. Please check network connectivity.") raise ValueError("Could not connect to the payment gateway. Please check network connectivity.")
if response.messages.resultCode == "Ok": if response.messages.resultCode == "Ok":
@@ -492,13 +516,6 @@ def charge_customer_profile(customer_profile_id: str, payment_profile_id: str, t
) )
controller = createTransactionController(createtransactionrequest) controller = createTransactionController(createtransactionrequest)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
# The response is returned directly to the router to be parsed there # The response is returned directly to the router to be parsed there

32
app/services/user_create.py
View File

@@ -15,27 +15,13 @@ from . import payment_service
from .. import schemas from .. import schemas
# Load Authorize.net credentials # Load Authorize.net credentials
from config import load_config, API_LOGIN_ID, TRANSACTION_KEY, VALIDATION_MODE, ENVIRONMENT
# Load Authorize.net environment variables
ApplicationConfig = load_config() ApplicationConfig = load_config()
# Set environment
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION': constants.environment = ENVIRONMENT
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
else:
constants.environment = constants.SANDBOX
constants.show_url_on_request = True
VALIDATION_MODE = "testMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
@@ -242,7 +228,7 @@ def create_user_account(db: Session, customer_id: int) -> dict:
} }
except Exception as e: except Exception as e:
logger.debug(f"Critical exception during user account creation for customer {customer_id}: {traceback.format_exc()}") logger.error(f"Critical exception during user account creation for customer {customer_id}: {traceback.format_exc()}")
db.rollback() db.rollback()
return { return {
"success": False, "success": False,
@@ -311,7 +297,7 @@ def refresh_customer_payment_profiles(db: Session, customer_id: int, auth_profil
) )
# Check test result # Check test result
_, _, test_reason = payment_service._parse_authnet_response(test_response) _, _, test_reason = payment_service.parse_authnet_response(test_response)
if "E00121" in str(test_reason): if "E00121" in str(test_reason):
cards_need_update.append(card) cards_need_update.append(card)
logger.debug(f"🔄 Card {card.id} has profile {card.auth_net_payment_profile_id} that EXISTS but is CORRUPTED - NEEDS RECREATION") logger.debug(f"🔄 Card {card.id} has profile {card.auth_net_payment_profile_id} that EXISTS but is CORRUPTED - NEEDS RECREATION")
@@ -373,7 +359,7 @@ def refresh_customer_payment_profiles(db: Session, customer_id: int, auth_profil
recreated_cards.append(card) recreated_cards.append(card)
logger.debug(f"✅ Successfully recreated payment profile {new_payment_profile_id} for card {card.id}") logger.debug(f"✅ Successfully recreated payment profile {new_payment_profile_id} for card {card.id}")
else: else:
logger.debug(f"❌ Failed to recreate payment profile for card {card.id} - no ID returned") logger.error(f"❌ Failed to recreate payment profile for card {card.id} - no ID returned")
except Exception as e: except Exception as e:
logger.debug(f"❌ Failed to recreate payment profile for card {card.id}: {str(e)}") logger.debug(f"❌ Failed to recreate payment profile for card {card.id}: {str(e)}")
@@ -383,7 +369,7 @@ def refresh_customer_payment_profiles(db: Session, customer_id: int, auth_profil
db.commit() db.commit()
logger.debug(f"✅ Successfully recreated and saved {len(recreated_cards)} payment profiles") logger.debug(f"✅ Successfully recreated and saved {len(recreated_cards)} payment profiles")
else: else:
logger.debug("❌ No payment profiles could be recreated - this is a critical failure") logger.error("❌ No payment profiles could be recreated - this is a critical failure")
return False return False
else: else:
logger.debug(f"🔄 All {len(cards_before)} cards have valid payment profile IDs") logger.debug(f"🔄 All {len(cards_before)} cards have valid payment profile IDs")

65
app/services/user_delete.py
View File

@@ -11,55 +11,18 @@ from authorizenet.constants import constants
from config import load_config from config import load_config
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
# Load Authorize.net credentials
from config import load_config, API_LOGIN_ID, TRANSACTION_KEY, VALIDATION_MODE, ENVIRONMENT
# Load Authorize.net credentials # Load Authorize.net credentials
ApplicationConfig = load_config() ApplicationConfig = load_config()
# Set environment
constants.environment = ENVIRONMENT
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
constants.environment = constants.PRODUCTION
VALIDATION_MODE = "liveMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
else:
constants.environment = constants.SANDBOX
constants.show_url_on_request = True
VALIDATION_MODE = "testMode"
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY
def _get_authnet_error_message(response): from .payment_service import _get_authnet_error_message
"""
Robust error parsing function that correctly handles the API's response format.
"""
if response is None:
return "No response from payment gateway."
try:
if hasattr(response, 'messages') and response.messages is not None:
if hasattr(response.messages, 'message'):
message_list = response.messages.message
if not isinstance(message_list, list):
message_list = [message_list]
if message_list:
msg = message_list[0]
code = msg.code if hasattr(msg, 'code') else 'Unknown'
text = msg.text if hasattr(msg, 'text') else 'No details provided.'
return f"Error {code}: {text}"
except Exception as e:
logger.debug(f"Error while parsing Auth.Net error message: {e}")
return "An unparsable error occurred with the payment gateway."
return "An unknown error occurred with the payment gateway."
def delete_user_account(db: Session, customer_id: int) -> dict: def delete_user_account(db: Session, customer_id: int) -> dict:
@@ -161,7 +124,7 @@ def delete_user_account(db: Session, customer_id: int) -> dict:
} }
except Exception as e: except Exception as e:
logger.debug(f"Critical exception during account deletion for customer {customer_id}: {traceback.format_exc()}") logger.error(f"Critical exception during account deletion for customer {customer_id}: {traceback.format_exc()}")
db.rollback() db.rollback()
return { return {
"success": False, "success": False,
@@ -192,13 +155,6 @@ def _delete_customer_profile(profile_id: str) -> bool:
) )
controller = deleteCustomerProfileController(request) controller = deleteCustomerProfileController(request)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
response = controller.getresponse() response = controller.getresponse()
@@ -247,13 +203,6 @@ def _delete_payment_profile(customer_profile_id: str, payment_profile_id: str) -
) )
controller = deleteCustomerPaymentProfileController(request) controller = deleteCustomerPaymentProfileController(request)
if ApplicationConfig.CURRENT_SETTINGS == 'PRODUCTION':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
elif ApplicationConfig.CURRENT_SETTINGS == 'LOCAL':
controller.setenvironment(constants.PRODUCTION)
controller.execute()
else:
controller.execute() controller.execute()
response = controller.getresponse() response = controller.getresponse()

28
app/utils.py Normal file
View File

@@ -0,0 +1,28 @@
"""
Utility functions shared across the EAMCO Authorize service.
"""
import re
def sanitize_input(text, max_len, allow_spaces=False, is_zip=False):
"""
Sanitize input string by removing invalid characters.
Args:
text: Input text/number/value
max_len: Maximum length of the output string
allow_spaces: If True, allows spaces in the output
is_zip: If True, allows hyphens (for ZIP+4)
Returns:
Sanitized string truncated to max_len
"""
if not text:
return ""
if is_zip:
pattern = r'[^a-zA-Z0-9-]'
else:
pattern = r'[^a-zA-Z0-9]' if not allow_spaces else r'[^a-zA-Z0-9\s]'
sanitized = re.sub(pattern, '', str(text))
return sanitized.strip()[:max_len]

45
config.py
View File

@@ -1,24 +1,55 @@
import os import os
# Authorize.net Configuration
from authorizenet.constants import constants
def load_config(mode=os.environ.get('MODE')): def load_config(mode=os.environ.get('MODE')):
"""
Load the application configuration based on the environment mode.
Sets up Authorize.net environment variables on the config object.
"""
config_class = None
try: try:
if mode == 'PRODUCTION': if mode == 'PRODUCTION':
from settings_prod import ApplicationConfig from settings_prod import ApplicationConfig
return ApplicationConfig config_class = ApplicationConfig
elif mode == 'LOCAL': elif mode == 'LOCAL':
from settings_local import ApplicationConfig from settings_local import ApplicationConfig
return ApplicationConfig config_class = ApplicationConfig
elif mode == 'DEVELOPMENT': elif mode == 'DEVELOPMENT':
from settings_dev import ApplicationConfig from settings_dev import ApplicationConfig
return ApplicationConfig config_class = ApplicationConfig
else: else:
pass # Default to dev if unknown mode
from settings_dev import ApplicationConfig
config_class = ApplicationConfig
except ImportError: except ImportError:
# Fallback
from settings_dev import ApplicationConfig from settings_dev import ApplicationConfig
return ApplicationConfig config_class = ApplicationConfig
# Set Authorize.net specific settings on the config class
if config_class.CURRENT_SETTINGS == 'PRODUCTION':
config_class.ENVIRONMENT = constants.PRODUCTION
config_class.VALIDATION_MODE = "liveMode"
elif config_class.CURRENT_SETTINGS == 'LOCAL':
config_class.ENVIRONMENT = constants.PRODUCTION
config_class.VALIDATION_MODE = "liveMode"
else:
config_class.ENVIRONMENT = constants.SANDBOX
config_class.VALIDATION_MODE = "testMode"
return config_class
# Load the configuration once
ApplicationConfig = load_config()
# Export variables for compatibility with imports in other files
ENVIRONMENT = ApplicationConfig.ENVIRONMENT
VALIDATION_MODE = ApplicationConfig.VALIDATION_MODE
API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID
TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY