Moibe's picture
Cleanup
28dcd97
raw
history blame
2.75 kB
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()
stripe.api_key = os.environ["STRIPE_KEY"]
# This is a terrible idea, only used for demo purposes!
app.state.stripe_customer_id = None
@app.get("/")
def start():
return {"Status":"Deployed"}
@app.get("/success")
async def success(request: Request):
return templates.TemplateResponse("success.html", {"request": request})
@app.get("/cancel")
async def cancel(request: Request):
return templates.TemplateResponse("cancel.html", {"request": request})
@app.post("/create-checkout-session")
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"]}
@app.post("/create-portal-session")
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}
@app.post("/webhook")
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)