Edwin Eames
97261f6c51
- 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
119 lines
4.7 KiB
Python
119 lines
4.7 KiB
Python
"""
|
|
CRUD operations for the EAMCO Authorize service.
|
|
"""
|
|
import logging
|
|
from sqlalchemy.orm import Session
|
|
from . import models, schemas
|
|
from .constants import TransactionStatus, TransactionType
|
|
from decimal import Decimal
|
|
from typing import Optional, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --- NEW CRUD FUNCTIONS FOR CIM ---
|
|
|
|
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()
|
|
|
|
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)
|
|
if db_customer:
|
|
db_customer.auth_net_profile_id = profile_id
|
|
db.commit()
|
|
db.refresh(db_customer)
|
|
return db_customer
|
|
|
|
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:]
|
|
# Schema guarantees YYYY-MM format
|
|
exp_year, exp_month = map(int, card_info.expiration_date.split('-'))
|
|
|
|
db_card = models.Card(
|
|
user_id=customer_id,
|
|
auth_net_payment_profile_id=payment_profile_id,
|
|
last_four_digits=last_four_digits,
|
|
type_of_card="Unknown", # Use a library like 'creditcard' to detect this from the number
|
|
expiration_year=exp_year,
|
|
expiration_month=exp_month
|
|
)
|
|
db.add(db_card)
|
|
db.commit()
|
|
db.refresh(db_card)
|
|
return db_card
|
|
|
|
# --- YOUR EXISTING CRUD FUNCTIONS (kept for completeness) ---
|
|
|
|
def get_customer(db: Session, customer_id: int):
|
|
return db.query(models.Customer).filter(models.Customer.id == customer_id).first()
|
|
|
|
def get_customer_cards(db: Session, customer_id: int):
|
|
"""
|
|
Get all cards for a specific customer.
|
|
"""
|
|
return db.query(models.Card).filter(models.Card.user_id == customer_id).all()
|
|
|
|
def get_customer_by_email(db: Session, email: str):
|
|
return db.query(models.Customer).filter(models.Customer.customer_email == email).first()
|
|
|
|
def get_customers(db: Session, skip: int = 0, limit: int = 100):
|
|
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) -> models.Transaction:
|
|
# Logic for amounts moved to routers/payment.py
|
|
# We simply save what is passed in the transaction object
|
|
preauthorize_amount = transaction.preauthorize_amount or Decimal("0.0")
|
|
charge_amount = transaction.charge_amount or Decimal("0.0")
|
|
|
|
db_transaction = models.Transaction(
|
|
preauthorize_amount=preauthorize_amount,
|
|
charge_amount=charge_amount,
|
|
transaction_type=transaction.transaction_type,
|
|
customer_id=customer_id,
|
|
status=status,
|
|
auth_net_transaction_id=auth_net_transaction_id,
|
|
service_id=transaction.service_id,
|
|
delivery_id=transaction.delivery_id,
|
|
auto_id=transaction.auto_id,
|
|
card_id=transaction.card_id,
|
|
payment_gateway=transaction.payment_gateway,
|
|
rejection_reason=transaction.rejection_reason
|
|
)
|
|
db.add(db_transaction)
|
|
db.commit()
|
|
db.refresh(db_transaction)
|
|
return db_transaction
|
|
|
|
def get_transaction_by_delivery_id(db: Session, delivery_id: int):
|
|
return db.query(models.Transaction).filter(
|
|
models.Transaction.delivery_id == delivery_id,
|
|
models.Transaction.transaction_type == TransactionType.AUTHORIZE,
|
|
models.Transaction.status == TransactionStatus.APPROVED
|
|
).first()
|
|
|
|
def get_transaction_by_auth_id(db: Session, auth_net_transaction_id: str):
|
|
return db.query(models.Transaction).filter(
|
|
models.Transaction.auth_net_transaction_id == auth_net_transaction_id
|
|
).first()
|
|
|
|
def get_transaction_by_auto_id(db: Session, auto_id: int):
|
|
return db.query(models.Transaction).filter(
|
|
models.Transaction.auto_id == auto_id,
|
|
models.Transaction.transaction_type.in_([TransactionType.AUTHORIZE, TransactionType.CAPTURE]),
|
|
models.Transaction.status == TransactionStatus.APPROVED
|
|
).first()
|
|
|
|
def update_transaction_for_capture(db: Session, auth_net_transaction_id: str, charge_amount: Decimal, status: int, rejection_reason: str = None):
|
|
transaction = db.query(models.Transaction).filter(models.Transaction.auth_net_transaction_id == auth_net_transaction_id).first()
|
|
if not transaction:
|
|
return None
|
|
|
|
transaction.charge_amount = charge_amount if status == TransactionStatus.APPROVED else Decimal("0.0")
|
|
transaction.transaction_type = TransactionType.CAPTURE
|
|
transaction.status = status
|
|
if rejection_reason:
|
|
transaction.rejection_reason = rejection_reason
|
|
|
|
db.commit()
|
|
db.refresh(transaction)
|
|
return transaction
|