88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
"""FastAPI backend for handling payments of membership fees."""
|
|
|
|
import logging
|
|
from typing import Annotated
|
|
import stripe
|
|
from fastapi import FastAPI, Form
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from admidio import settings
|
|
from admidio.model import AdmidioMemberFee
|
|
|
|
LOG = logging.getLogger()
|
|
handler = logging.StreamHandler()
|
|
LOG.addHandler(handler)
|
|
LOG.setLevel(logging.DEBUG)
|
|
|
|
hostname = settings.get_domain()
|
|
db_settings = settings.get_db_settings()
|
|
stripe_key = settings.get_stripe_key()
|
|
stripe.api_key = stripe_key
|
|
app = FastAPI()
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
|
|
def render_dynamic(title: str, message: str) -> HTMLResponse:
|
|
"""Render the dynamic page."""
|
|
filename = "dynamic"
|
|
template = open(f"static/{filename}.html", "r").read()
|
|
html = template.format(title=title, message=message)
|
|
return HTMLResponse(html)
|
|
|
|
|
|
def render_static(filename: str) -> HTMLResponse:
|
|
"""Render a static page."""
|
|
html = open(f"static/{filename}.html", "r").read()
|
|
return HTMLResponse(html)
|
|
|
|
|
|
@app.get("/")
|
|
def index() -> HTMLResponse:
|
|
"""Serve index page."""
|
|
return render_static("index")
|
|
|
|
|
|
@app.post("/checkout", response_model=None)
|
|
def create_checkout(email: Annotated[str, Form()]) -> RedirectResponse | HTMLResponse:
|
|
"""Create a checkout session."""
|
|
LOG.info("Got checkout request.")
|
|
member_fee = AdmidioMemberFee.lookup_email(email, db_settings)
|
|
if not member_fee:
|
|
return render_dynamic("Error", "Could not find a member with this email.")
|
|
LOG.debug("User data: %r", member_fee.user_data)
|
|
if member_fee.has_paid:
|
|
return render_dynamic(
|
|
"Already paid", "You have already paid your membership for this year."
|
|
)
|
|
stripe_price = settings.get_stripe_price()
|
|
|
|
session = stripe.checkout.Session.create(
|
|
line_items=[
|
|
{
|
|
"price": stripe_price,
|
|
"quantity": 1,
|
|
}
|
|
],
|
|
mode="payment",
|
|
success_url=f"{hostname}/process-payment/{member_fee.user_id}?session_id={{CHECKOUT_SESSION_ID}}",
|
|
cancel_url=f"{hostname}/static/failure.html",
|
|
)
|
|
if session.url:
|
|
return RedirectResponse(session.url, 303)
|
|
return render_static("failure")
|
|
|
|
|
|
@app.get("/process-payment/{user_id}", response_model=None)
|
|
def process_payment(user_id: int, session_id: str) -> HTMLResponse:
|
|
"""Process payment."""
|
|
session = stripe.checkout.Session.retrieve(session_id)
|
|
status = session.payment_status
|
|
if status == "paid":
|
|
member_fee = AdmidioMemberFee(user_id, db_settings)
|
|
member_fee.register_payment()
|
|
return render_static("success")
|
|
|
|
return render_static("failure")
|