100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
## File: your_app/crud.py
|
|
|
|
from sqlalchemy.orm import Session
|
|
from . import models, schemas
|
|
from decimal import Decimal
|
|
|
|
# --- NEW CRUD FUNCTIONS FOR CIM ---
|
|
|
|
def get_card_by_id(db: Session, card_id: int):
|
|
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):
|
|
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):
|
|
last_four_digits = card_info.card_number[-4:]
|
|
try:
|
|
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(
|
|
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_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):
|
|
# Using your existing logic for saving amounts
|
|
preauthorize_amount = transaction.preauthorize_amount if status == 0 else Decimal("0.0")
|
|
charge_amount = transaction.charge_amount if (transaction.transaction_type != 1 and status == 0) else 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,
|
|
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 == 1,
|
|
models.Transaction.status == 0
|
|
).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 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 == 0 else Decimal("0.0")
|
|
transaction.transaction_type = 2
|
|
transaction.status = status
|
|
if rejection_reason:
|
|
transaction.rejection_reason = rejection_reason
|
|
|
|
db.commit()
|
|
db.refresh(transaction)
|
|
return transaction |