import os from typing import Annotated import qrcode import qrcode.image.svg from fastapi import FastAPI, Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from mollie.api.client import Client app = FastAPI(docs_url=None, redoc_url=None) app.mount("/static", StaticFiles(directory="resources/static"), name="static") templates = Jinja2Templates(directory="resources") public_url = os.environ.get("PUBLIC_URL", "http://localhost:8000").rstrip("/") mollie_apikey = os.environ.get("MOLLIE_API_KEY", "test_test") mollie_client = Client() mollie_client.set_api_key(mollie_apikey) @app.get("/", response_class=HTMLResponse) async def welcome(request: Request): return templates.TemplateResponse(request=request, name="welcome.html.j2") @app.post("/") async def pay(request: Request, amount: Annotated[str, Form()]): # Normalize input. if "," in amount: amount = amount.replace(",", ".") if "." not in amount: amount = f"{amount}.00" # Conversion. [i, f] = amount.split(".") cents = int(i) * 100 + int(f) # Sanity checks. if cents < 10_00 or 150_00 < cents: raise HTTPException(status_code=400, detail=f"invalid amount: {amount}") payment = mollie_client.payments.create( { "amount": { "currency": "EUR", "value": f"{cents // 100}.{cents % 100:02}", }, "description": "Bitlair bar tegoed", "redirectUrl": f"{public_url}/return", "metadata": {"revbank_status": "unspent"}, } ) mollie_client.payments.update( payment.id, {"redirectUrl": f"{payment.redirect_url}/{payment.id}"} ) return RedirectResponse(payment.checkout_url, status_code=302) @app.get("/return/{id}", response_class=HTMLResponse) async def return_(request: Request, id: str): qr = qrcode.make(id, image_factory=qrcode.image.svg.SvgPathImage) return templates.TemplateResponse( request=request, name="return.html.j2", context={ "code": id, "qr_svg": qr.to_string(encoding="unicode"), }, ) def not_ok(msg: str): return {"ok": False, "message": msg} def ok(**kwargs): return {"ok": True, **kwargs} @app.post("/revbank_plugin_backend") async def rb_backend( id: Annotated[str, Form()], action: Annotated[str | None, Form()] = None ): payment = mollie_client.payments.get(id) if not payment.is_paid(): return not_ok(f"payment {payment.status}") assert payment.amount["currency"] == "EUR" rb_status = payment.metadata.get("revbank_status", None) assert rb_status is not None # Called to reserve a payment in a Revbank transction. if action is None: if rb_status != "unspent": return not_ok("already spent") mollie_client.payments.update( payment.id, {"metadata": {"revbank_status": "pending"}}, ) if mollie_apikey.startswith("test_"): return ok(amount="0.00", test_amount=payment.amount["value"]) return ok(amount=payment.amount["value"]) # Called when a pending Revbank transaction is aborted by the user. if action == "abort": if rb_status != "pending": return not_ok(f"can't abort non-pending, {rb_status}") mollie_client.payments.update( payment.id, {"metadata": {"revbank_status": "unspent"}}, ) return ok() # Called when Revbank finishes a pending transaction. if action == "finalize": mollie_client.payments.update( payment.id, {"metadata": {"revbank_status": "spent"}}, ) return ok() raise HTTPException(status_code=400, detail=f"invalid action {action}")