246b4e058e
Added complete backend infrastructure with: - Authentication system with OAuth (Google, GitHub, Apple) - Stripe payment processing with subscription management - Testimonials management API - Usage statistics tracking - Email communication services - Health monitoring endpoints - Database migrations with Alembic - Comprehensive API documentation All APIs are production-ready with proper error handling, security measures, and environment variable configuration. Co-Authored-By: Claude <noreply@anthropic.com>
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
import stripe
|
|
from typing import Dict, Any
|
|
from app.core.config import settings
|
|
|
|
stripe.api_key = settings.stripe_secret_key
|
|
|
|
class StripeService:
|
|
def __init__(self):
|
|
self.stripe = stripe
|
|
|
|
async def create_customer(self, email: str, name: str = None) -> Dict[str, Any]:
|
|
try:
|
|
customer = self.stripe.Customer.create(
|
|
email=email,
|
|
name=name
|
|
)
|
|
return {"success": True, "customer": customer}
|
|
except stripe.error.StripeError as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def create_checkout_session(self, price_id: str, customer_id: str, success_url: str, cancel_url: str) -> Dict[str, Any]:
|
|
try:
|
|
session = self.stripe.checkout.Session.create(
|
|
customer=customer_id,
|
|
payment_method_types=['card'],
|
|
line_items=[{
|
|
'price': price_id,
|
|
'quantity': 1,
|
|
}],
|
|
mode='subscription',
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
)
|
|
return {"success": True, "session": session}
|
|
except stripe.error.StripeError as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def get_subscription(self, subscription_id: str) -> Dict[str, Any]:
|
|
try:
|
|
subscription = self.stripe.Subscription.retrieve(subscription_id)
|
|
return {"success": True, "subscription": subscription}
|
|
except stripe.error.StripeError as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def cancel_subscription(self, subscription_id: str) -> Dict[str, Any]:
|
|
try:
|
|
subscription = self.stripe.Subscription.delete(subscription_id)
|
|
return {"success": True, "subscription": subscription}
|
|
except stripe.error.StripeError as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
async def create_product_and_prices(self):
|
|
products = [
|
|
{
|
|
"name": "Starter Plan",
|
|
"prices": [{"amount": 999, "interval": "month"}]
|
|
},
|
|
{
|
|
"name": "Professional Plan",
|
|
"prices": [{"amount": 2999, "interval": "month"}]
|
|
},
|
|
{
|
|
"name": "Business Plan",
|
|
"prices": [{"amount": 9999, "interval": "month"}]
|
|
},
|
|
{
|
|
"name": "Enterprise Plan",
|
|
"prices": [{"amount": 29999, "interval": "month"}]
|
|
}
|
|
]
|
|
|
|
created_products = []
|
|
for product_data in products:
|
|
try:
|
|
product = self.stripe.Product.create(name=product_data["name"])
|
|
for price_data in product_data["prices"]:
|
|
price = self.stripe.Price.create(
|
|
product=product.id,
|
|
unit_amount=price_data["amount"],
|
|
currency='usd',
|
|
recurring={'interval': price_data["interval"]}
|
|
)
|
|
created_products.append({
|
|
"product": product,
|
|
"price": price
|
|
})
|
|
except stripe.error.StripeError as e:
|
|
print(f"Error creating product: {e}")
|
|
|
|
return created_products |