a8b6d5b972
- Renamed 'metadata' column to 'payment_metadata' in Payment model - Updated database migration to use new column name - Updated StripeService to use payment_metadata field - Fixed application startup issue caused by reserved attribute name
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
import stripe
|
|
import os
|
|
from typing import Dict, Any, Optional
|
|
from sqlalchemy.orm import Session
|
|
from app.models.payment import Payment
|
|
import json
|
|
|
|
|
|
class StripeService:
|
|
def __init__(self):
|
|
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
|
|
self.webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET")
|
|
|
|
def create_payment_intent(
|
|
self,
|
|
amount: int,
|
|
currency: str = "usd",
|
|
customer_email: Optional[str] = None,
|
|
customer_name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
) -> stripe.PaymentIntent:
|
|
"""Create a Stripe payment intent"""
|
|
payment_intent_data = {
|
|
"amount": amount,
|
|
"currency": currency,
|
|
"automatic_payment_methods": {"enabled": True},
|
|
}
|
|
|
|
if customer_email:
|
|
payment_intent_data["receipt_email"] = customer_email
|
|
|
|
if description:
|
|
payment_intent_data["description"] = description
|
|
|
|
if metadata:
|
|
payment_intent_data["metadata"] = metadata
|
|
|
|
return stripe.PaymentIntent.create(**payment_intent_data)
|
|
|
|
def confirm_payment_intent(self, payment_intent_id: str) -> stripe.PaymentIntent:
|
|
"""Confirm a payment intent"""
|
|
return stripe.PaymentIntent.confirm(payment_intent_id)
|
|
|
|
def retrieve_payment_intent(self, payment_intent_id: str) -> stripe.PaymentIntent:
|
|
"""Retrieve a payment intent"""
|
|
return stripe.PaymentIntent.retrieve(payment_intent_id)
|
|
|
|
def create_customer(
|
|
self, email: str, name: Optional[str] = None
|
|
) -> stripe.Customer:
|
|
"""Create a Stripe customer"""
|
|
customer_data = {"email": email}
|
|
if name:
|
|
customer_data["name"] = name
|
|
return stripe.Customer.create(**customer_data)
|
|
|
|
def save_payment_to_db(
|
|
self,
|
|
db: Session,
|
|
payment_intent: stripe.PaymentIntent,
|
|
customer_email: Optional[str] = None,
|
|
customer_name: Optional[str] = None,
|
|
) -> Payment:
|
|
"""Save payment data to database"""
|
|
payment = Payment(
|
|
stripe_payment_intent_id=payment_intent.id,
|
|
amount=payment_intent.amount / 100, # Convert from cents
|
|
currency=payment_intent.currency,
|
|
status=payment_intent.status,
|
|
customer_email=customer_email,
|
|
customer_name=customer_name,
|
|
description=payment_intent.description,
|
|
payment_metadata=json.dumps(payment_intent.metadata)
|
|
if payment_intent.metadata
|
|
else None,
|
|
)
|
|
db.add(payment)
|
|
db.commit()
|
|
db.refresh(payment)
|
|
return payment
|
|
|
|
def update_payment_status(
|
|
self, db: Session, payment_intent_id: str, status: str
|
|
) -> Optional[Payment]:
|
|
"""Update payment status in database"""
|
|
payment = (
|
|
db.query(Payment)
|
|
.filter(Payment.stripe_payment_intent_id == payment_intent_id)
|
|
.first()
|
|
)
|
|
if payment:
|
|
payment.status = status
|
|
db.commit()
|
|
db.refresh(payment)
|
|
return payment
|
|
|
|
def verify_webhook_signature(self, payload: bytes, sig_header: str) -> bool:
|
|
"""Verify Stripe webhook signature"""
|
|
try:
|
|
stripe.Webhook.construct_event(payload, sig_header, self.webhook_secret)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
except stripe.error.SignatureVerificationError:
|
|
return False
|
|
|
|
def handle_webhook_event(self, db: Session, event: Dict[str, Any]) -> bool:
|
|
"""Handle Stripe webhook events"""
|
|
if event["type"] == "payment_intent.succeeded":
|
|
payment_intent = event["data"]["object"]
|
|
self.update_payment_status(db, payment_intent["id"], "succeeded")
|
|
|
|
# Mark webhook as processed
|
|
payment = (
|
|
db.query(Payment)
|
|
.filter(Payment.stripe_payment_intent_id == payment_intent["id"])
|
|
.first()
|
|
)
|
|
if payment:
|
|
payment.is_webhook_processed = True
|
|
db.commit()
|
|
return True
|
|
|
|
elif event["type"] == "payment_intent.payment_failed":
|
|
payment_intent = event["data"]["object"]
|
|
self.update_payment_status(db, payment_intent["id"], "failed")
|
|
return True
|
|
|
|
return False
|