"""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")