Spaces:
Running
Running
import os | |
import stripe | |
import uvicorn | |
from fastapi import FastAPI, Request, Header | |
from fastapi.staticfiles import StaticFiles | |
from fastapi.templating import Jinja2Templates | |
import time | |
app = FastAPI() | |
templates = Jinja2Templates(directory="templates") | |
app.mount("/static", StaticFiles(directory="static"), name="static") | |
stripe.api_key = os.environ["STRIPE_KEY"] | |
# This is a terrible idea, only used for demo purposes! | |
app.state.stripe_customer_id = None | |
def index(request: Request): | |
return templates.TemplateResponse("index.html", {"request": request, "hasCustomer": app.state.stripe_customer_id is not None}) | |
async def success(request: Request): | |
return templates.TemplateResponse("success.html", {"request": request}) | |
async def cancel(request: Request): | |
return templates.TemplateResponse("cancel.html", {"request": request}) | |
async def create_checkout_session(request: Request): | |
data = await request.json() | |
if not app.state.stripe_customer_id: | |
customer = stripe.Customer.create( | |
description="Demo customer", | |
) | |
app.state.stripe_customer_id = customer["id"] | |
checkout_session = stripe.checkout.Session.create( | |
customer=app.state.stripe_customer_id, | |
success_url="http://localhost:8000/success?session_id={CHECKOUT_SESSION_ID}", | |
cancel_url="http://localhost:8000/cancel", | |
payment_method_types=["card"], | |
mode="subscription", | |
line_items=[{ | |
"price": data["priceId"], | |
"quantity": 1 | |
}], | |
) | |
return {"sessionId": checkout_session["id"]} | |
async def create_portal_session(): | |
session = stripe.billing_portal.Session.create( | |
customer=app.state.stripe_customer_id, | |
return_url="http://localhost:8000" | |
) | |
return {"url": session.url} | |
async def webhook_received(request: Request, stripe_signature: str = Header(None)): | |
webhook_secret = os.environ["STRIPE_WEBHOOK_SECRET"] | |
print("Entré al webhook 182...") | |
time.sleep(1) | |
data = await request.body() | |
try: | |
event = stripe.Webhook.construct_event( | |
payload=data, | |
sig_header=stripe_signature, | |
secret=webhook_secret | |
) | |
event_data = event['data'] | |
except Exception as e: | |
return {"error": str(e)} | |
event_type = event['type'] | |
print("Voy a imprimir el event type:") | |
print(event_type) | |
time.sleep(1) | |
if event_type == 'checkout.session.completed': | |
print('checkout session completed') | |
elif event_type == 'invoice.paid': | |
print('invoice paid') | |
elif event_type == 'invoice.payment_failed': | |
print('invoice payment failed') | |
else: | |
print(f'unhandled event: {event_type}') | |
return {"status": "success"} | |
if __name__ == '__main__': | |
uvicorn.run("app:app", reload=True) |