import logging import os from contextlib import asynccontextmanager from pathlib import Path from typing import Literal from urllib.parse import urljoin, urlparse import stripe from asgi_csrf import asgi_csrf from dotenv import load_dotenv from fastapi import BackgroundTasks, FastAPI, Form, HTTPException, Request from fastapi.encoders import jsonable_encoder from fastapi.exceptions import RequestValidationError from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.exceptions import HTTPException as StarletteHTTPException from starsessions import CookieStore, SessionAutoloadMiddleware, SessionMiddleware from pydantic import ValidationError, EmailStr import pycountry from app.cart import ( calculate_quantity_requested_by_user, clear_cart, initialize_cart, product_exist_in_cart, update_cart, update_cart_meta, create_product_list, ) from app.db import ( check_product_availability, export_order_to_xlsx, fetch_order_product_list, get_products_by_category, is_order_id_open, prepare_customer_order_data, save_shipping_info, update_product_inventory, update_status_order, write_variation_id, get_variation_id, write_product_id_db, get_products_id_db, ) from app.email import send_email from app.parser import md, read_dir, read_file from app.payment import ( calculate_shipping_cost, prepare_shipping_rate, generate_payment_link, ipn_signature_check, prepare_stripe_data, ) from app.read_settings import read_settings from app.schema import CartSession, CartUpdate, NowPaymentsInvoice, ShippingInfo, ProductShippingMeta from app.template import ( make_product_selected, prepare_checkout, prepare_email_order, prepare_product, prepare_related_products, prepare_sets_for_menu, to_srcset, assets_hashing, make_slugify ) from app.form_validation import form_validation_checkout_session load_dotenv(".env") logging_level = logging.INFO main_logger = logging.getLogger() main_logger.setLevel(logging_level) formatter = logging.Formatter("%(asctime)s %(levelname)s: %(name)s => %(message)s") # Set up a stream handler to log to the console stream_handler = logging.StreamHandler() stream_handler.setLevel(logging_level) stream_handler.setFormatter(formatter) file_handler = logging.FileHandler("logs.log") file_handler.setLevel(logging_level) file_handler.setFormatter(formatter) # Add handlers to logger main_logger.addHandler(stream_handler) main_logger.addHandler(file_handler) @asynccontextmanager async def lifespan(app: FastAPI): """ On startup of application we check that all the product variations have their unique IDs """ products_list = read_dir( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, exclude_doc="products", tree="products", ) products_data = get_products_id_db(settings) for single_product in products_list: write_variation_id(single_product, settings) write_product_id_db(single_product, settings, products_data) yield app = FastAPI(docs_url=None, redoc_url=None, lifespan=lifespan) # -- read settings settings = read_settings("settings.toml") # -- mount static files app.mount( "/static", StaticFiles(directory=Path(__file__).parent.parent / "static"), name="static", ) # mount media folder from git content repo app.mount("/media", StaticFiles(directory=settings.git_repo), name="media") # -- setup templates templates = Jinja2Templates(directory=Path(__file__).parent / "templates") templates.env.filters["md"] = md templates.env.filters["to_srcset"] = to_srcset templates.env.filters["assets_hashing"] = assets_hashing templates.env.filters["slugify"] = make_slugify templates.env.globals.update(CACHE=os.getenv('CACHE')) # -- add middlewares for CSRF and Session app.add_middleware( asgi_csrf, signing_secret=os.getenv("CSRF_SECRET_KEY"), always_protect={"/", "/products", "/checkout"}, ) session_store = CookieStore(secret_key=os.getenv("SESSION_SECRET_KEY")) app.add_middleware(SessionAutoloadMiddleware) app.add_middleware( SessionMiddleware, store=session_store, lifetime=3600 * 24 * 14, cookie_https_only=os.getenv("ENV") == "dev", ) # -- load Stripe and set API key if os.getenv("ENV") == "production": stripe.api_key = os.getenv("STRIPE_SECRET_KEY") else: stripe.api_key = os.getenv("STRIPE_TEST_SECRET_KEY") @app.get("/", response_class=HTMLResponse) async def root(request: Request): """The Main view of the website. It gives access to all the pages.""" session_data = request.session initialize_cart(session_data, settings.currency.default) cart = CartSession(**request.session["cart"]) doc = read_file( settings.git_repo, "index", settings.document_match, settings.block_types, "", ) categories = get_products_by_category(settings) sets = prepare_sets_for_menu(settings) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) response = templates.TemplateResponse( request, "index.html", { "request": request, "nav": None, "doc": doc, "categories": categories, "sets": sets, "site": site, "cart": cart, }, ) return response @app.get("/products", response_class=RedirectResponse, status_code=301) async def products(): """ The products view does not exist, so we return the user to the main view. """ return "/" @app.get("/products/{product_id}", response_class=HTMLResponse) async def product( request: Request, product_id: str, js: bool = False, size: str | None = None, style: str | None = None, quantity: int = 1, ): """The product view displaying the content of the selected product PRODUCT ID. By default we check if the product exists in quantity = 1, unless a different value is explicitly pass to the GET request. """ session_data = request.session initialize_cart(session_data, settings.currency.default) cart = CartSession(**request.session["cart"]) tree = "products" doc = read_file( settings.git_repo, product_id, settings.document_match, settings.block_types, tree, ) quantity_in_cart = calculate_quantity_requested_by_user( session_data["cart"]["items"], doc.meta.product_id, size=size, style=style ) # For the product page, we want to check if adding 1 more would exceed stock quantity_requested_by_user = quantity_in_cart + 1 triplet = check_product_availability( doc, product_id, quantity_requested_by_user, size, style, settings ) is_product_available, is_product_selectable, inventory_amount = triplet if js: product_selected = make_product_selected(size, style, quantity) json_response = { "is_product_available": is_product_available, "is_product_selectable": is_product_selectable, "product_selected": product_selected, } return JSONResponse(content=json_response, status_code=200) else: doc.blocks = prepare_product(doc) size_guide = None if doc.meta.options: size_guide = read_file( settings.git_repo, "products", settings.document_match, settings.block_types, None, ) products = read_dir( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, exclude_doc=doc.meta.path, tree="products", ) related_products = prepare_related_products(products, product_id) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) response = templates.TemplateResponse( request, "product.html", { "request": request, "nav": None, "doc": doc, "size_guide": size_guide, "related_products": related_products, "is_product_available": is_product_available, "is_product_selectable": is_product_selectable, "product_form": { "size": size, "style": style, "quantity": quantity, }, "site": site, "cart": cart, }, ) return response @app.get("/pages", response_class=RedirectResponse, status_code=301) async def products(): """ The pages view does not exist, so we return the user to the main view. """ return "/" @app.get("/pages/{page_id}", response_class=HTMLResponse) async def page(request: Request, page_id: str): """The Page view displaying the content of the selected PAGE ID.""" tree = "pages" doc = read_file( settings.git_repo, page_id, settings.document_match, settings.block_types, tree, ) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) response = templates.TemplateResponse( request, "page.html", { "request": request, "nav": None, "doc": doc, "site": site, }, ) return response @app.get("/checkout", response_class=HTMLResponse) async def checkout(request: Request): """The Checkout view (GET, read-only).""" filename = request.url.path[1:] doc = read_file( settings.git_repo, filename, settings.document_match, settings.block_types ) session_data = request.session initialize_cart(session_data, settings.currency.default) update_cart_meta( session_data["cart"], session_data["cart"]["meta"]["shipping"], settings ) cart = CartSession(**request.session["cart"]) checkout_data = prepare_checkout(cart, settings, doc) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) response = templates.TemplateResponse( request, "checkout.html", { "request": request, "nav": None, "data": checkout_data, "site": site, "cart": cart, }, ) return response @app.post("/checkout") async def checkout_update( request: Request, js: bool = False, page: Literal["product", "checkout"] = Form(...), redirect_with_items: bool | None = Form(None), operation: Literal["add", "remove", "delete", "manual"] = Form(...), product_id: str = Form(...), product_path: str = Form(...), price: int = Form(...), quantity: int = Form(...), size: Literal["s", "m", "l", "xl", "xxl"] | None = Form(None), style: str | None = Form(None), ): """The Checkout view, (POST, read-write). This view updates (add / remove) the user's session cart. If the item is already present in the cart, it increases or decreases the item quantity. Else, it will be created or removed accordingly. It will also check if any selected item in the cart is still available or it has sold out meanwhile. Morever, it displays the Checkout view. """ main_logger.info(f"POST /checkout: js={js}, page={page}, product_id={product_id}, size={size}, style={style}, quantity={quantity}") session_data = request.session initialize_cart(session_data, settings.currency.default) shipping_amount = session_data["cart"]["meta"]["shipping"] if operation == "manual": selected_product_id = session_data["cart"]["items"][product_id] price = quantity * selected_product_id["price"] tree = "products" product_filename = Path(product_path).stem selected_product = read_file( settings.git_repo, product_filename, settings.document_match, settings.block_types, tree, ) product_variation_id = get_variation_id(selected_product, size, style) form = CartUpdate( operation=operation, product_id=product_id, options={"size": size, "style": style, "product_variation_id": product_variation_id, "weight": selected_product.meta.weight, }, price=price, ) quantity_in_cart = calculate_quantity_requested_by_user( session_data["cart"]["items"], product_id, size=size, style=style ) quantity_requested_by_user = quantity_in_cart + quantity main_logger.info(f"Calling check_product_availability with: product_id={product_id}, quantity_requested_by_user={quantity_requested_by_user}, size={size}, style={style}") triplet = check_product_availability( selected_product, product_path, quantity_requested_by_user, size, style, settings, ) is_product_available, is_product_selectable, inventory_amount = triplet main_logger.info(f"check_product_availability returned: is_product_available={is_product_available}, is_product_selectable={is_product_selectable}, inventory_amount={inventory_amount}") product_exists = product_exist_in_cart( product_id, size, style, product_variation_id, session_data["cart"]["items"] ) update_cart( session_data["cart"], form, product_exists, shipping_amount, quantity, inventory_amount, settings, ) cart = CartSession(**request.session["cart"]) if js: # -- return JSON res for js-handled form update if page == "product": cart_breakdown = jsonable_encoder(cart.meta) product_selected = make_product_selected(size, style, quantity) json_response = { "cart_breakdown": cart_breakdown, "is_product_available": is_product_available, "is_product_selectable": is_product_selectable, "product_selected": product_selected, } elif page == "checkout": filename = request.url.path[1:] doc = read_file( settings.git_repo, filename, settings.document_match, settings.block_types, ) checkout_data = prepare_checkout( cart, settings, doc, js_update=False ) cart_items = [ jsonable_encoder(item) for item in checkout_data["checkout_items"] ] cart_breakdown = jsonable_encoder(cart.meta) json_response = { "cart_items": cart_items, "cart_breakdown": cart_breakdown, } main_logger.info(f"AJAX response: {json_response}") return JSONResponse(content=json_response, status_code=200) else: redirect_URL = urljoin( request.headers["referer"], urlparse(request.headers["referer"]).path ) redirect_URL = f"{redirect_URL}" if redirect_with_items: URI_options = [] param_options = {"size": size, "style": style, "quantity": quantity} for idx, option in enumerate(param_options.keys()): if param_options[option]: param = f"{option}={param_options[option]}" URI_options.append(param) if len(URI_options) > 0: URI_params = "&".join(URI_options) redirect_URL = f"{redirect_URL}?{URI_params}" else: redirect_URL = f"{redirect_URL}" return RedirectResponse(redirect_URL, status_code=303) @app.post("/checkout/shipping") async def checkout_shipping(request: Request, js: bool = False, country: str = Form(...), weight: int = Form(...)): """ This view returns the region to which the given submitted country belongs to. This is used in /checkout to update shipping costs in real time. """ countries = [country.name for country in pycountry.countries] if ( country not in countries or weight <= 0 ): raise HTTPException(status_code=403, detail="Shipping info are incorrect.") shipping_rate = calculate_shipping_cost(country, weight, settings.currency.default) if shipping_rate: # update cart with new shipping costs session_data = request.session initialize_cart(session_data, settings.currency.default) update_cart_meta(session_data["cart"], shipping_rate.price, settings) cart = CartSession(**request.session['cart']) if js: return cart.meta else: return RedirectResponse(f"{request.base_url}checkout", status_code=303) raise HTTPException(status_code=403, detail="Shipping info are missing.") @app.post("/stripe-checkout-session", response_class=RedirectResponse, status_code=303) async def create_checkout_session( request: Request, order_id: str = Form(...), weight: int = Form(...), email: EmailStr = Form(...), first_name: str | None = Form(None), last_name: str = Form(...), address: str = Form(...), address_no: str = Form(...), address_extra: str | None = Form(None), postal_code: str = Form(...), city: str = Form(...), country: str = Form(...), phone_number: str = Form(None), note: str | None = Form(None), ): """Stripe-based checkout session. Prepare order data and create a new Stripe Checkout session. Handle gracefully in case of errors, or if order exists already in local-db and has not `status: Open`. """ # -- initial form-data validation form_data = await request.form() try: form = form_validation_checkout_session(form_data) except ValidationError as e: for error in e.errors(): main_logger.error(f"form-validation-stripe-checkout-session => {error}") # -- check if order-id is unique orders_settings = { "git_repo": settings.git_repo, "document_match": settings.document_match, "block_types": settings.block_types, } if not is_order_id_open(orders_settings, form.order_id): main_logger.error("stripe-checkout error: Order ID is wrong.") raise HTTPException(status_code=422, detail="Order ID is wrong.") shipping_info = { "email": form.email, "first_name": form.first_name, "last_name": form.last_name, "address": form.address, "address_no": form.address_no, "address_extra": form.address_extra, "postal_code": form.postal_code, "city": form.city, "country": form.country, "phone_number": form.phone_number, "note": form.note, } shipping_info = ShippingInfo(**shipping_info) session_data = request.session initialize_cart(session_data, settings.currency.default) if "cart" not in session_data and len(session_data["cart"].keys()) == 0: main_logger.error("stripe-checkout error: Cart is empty.") raise HTTPException(status_code=403, detail="Cart is empty.") cart = CartSession(**request.session["cart"]) try: line_items = prepare_stripe_data(cart) shipping_rate = prepare_shipping_rate(shipping_info.country, form.weight, settings.currency.default) if shipping_rate is None: main_logger.error("stripe-checkout error: Shipping info are missing.") raise HTTPException(status_code=403, detail="Shipping info are missing.") checkout_session = stripe.checkout.Session.create( mode="payment", line_items=line_items, shipping_options=[shipping_rate], customer_email=shipping_info.email, client_reference_id=form.order_id, success_url=f"{request.base_url}checkout/success", cancel_url=f"{request.base_url}checkout", ) product_list = create_product_list(settings, cart.items) # -- save transaction to local-db customer_order = prepare_customer_order_data( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, checkout_session["id"], product_list, "Stripe", form.order_id, cart, shipping_info, ) save_shipping_info( customer_order, settings.git_repo, settings.local_db.filepath ) # -- redirect to Stripe Checkout page return checkout_session.url except stripe.error.CardError as e: main_logger.error( f"stripe-checkout error: A payment error occurred: {e.user_message}" ) raise HTTPException(status_code=400, detail="Stripe payment error.") except stripe.error.InvalidRequestError as e: main_logger.error(f"stripe-checkout error: An invalid request occurred. => {e}") raise HTTPException( status_code=400, detail="Stripe payment error (invalid request)." ) except Exception as e: main_logger.error( f"stripe-checkout error: Another problem occurred, maybe unrelated to Stripe. {e}" ) raise HTTPException(status_code=400, detail="Stripe payment error.") @app.post("/stripe-webhook", response_class=JSONResponse) async def stripe_webhook(request: Request, background_tasks: BackgroundTasks): """Stripe webhook endpoint to receive updates from Stripe's checkout operations. Send email after successful order confirmation. """ main_logger.info("stripe-webhook => ...") try: event = None payload = await request.body() sig_header = request.headers["stripe-signature"] try: if os.getenv("ENV") == "production": endpoint_secret = os.getenv("STRIPE_ENDPOINT_SECRET") else: endpoint_secret = os.getenv("STRIPE_TEST_ENDPOINT_SECRET") event = stripe.Webhook.construct_event(payload, sig_header, endpoint_secret) except ValueError as e: main_logger.error( f"stripe-webhook error: Webhook error while parsing basic request. {e}" ) raise HTTPException(status_code=400, detail="Stripe payment error.") except stripe.error.SignatureVerificationError as e: main_logger.error( f"stripe-webhook error: Webhook signature verification failed. {e}" ) raise HTTPException(status_code=400, detail="Stripe payment error.") if event: checkout = event["data"]["object"] if "client_reference_id" in checkout: try: status_res = update_status_order( event["type"], checkout["client_reference_id"], settings.git_repo, settings.local_db.filepath, ) if status_res is False: main_logger.error("stripe-webhook error: order could not be updated.") # Don't raise an exception, just log and continue #raise HTTPException( # status_code=400, detail="Order could not be updated." #) # We still want to try updating inventory and sending email except KeyError as e: main_logger.error(f"stripe-webhook error: Missing key in order data: {e}") # Continue processing - we want to still try to send the email except Exception as e: main_logger.error(f"stripe-webhook error: Unexpected error updating order: {e}") # Continue processing if event["type"] in [ "checkout.session.completed", "checkout.session.async_payment_succeeded", ]: order_id = checkout["client_reference_id"] product_list = fetch_order_product_list( order_id, settings.git_repo, settings.local_db.filepath ) if product_list: for product_info in product_list: # We need to read the product file and get the variation ID try: product = read_file( settings.git_repo, f"products/{product_info.filename}", settings.document_match, settings.block_types, ) # Get the variation ID based on size and style product_variation_id = get_variation_id(product, product_info.size, product_info.style) update_product_inventory( product_info.filename, product_info.quantity, product_variation_id, settings, ) except Exception as e: main_logger.error(f"stripe-webhook error: Failed to update inventory: {e}") # Continue processing other products # -- send email try: email_settings = { "git_repo": settings.git_repo, "document_match": settings.document_match, "block_types": settings.block_types, } # It's possible the order was just created and not all data is available yet # The prepare_email_order function should handle this gracefully try: order = prepare_email_order(order_id, email_settings) await send_email(email_settings, order, background_tasks) except TypeError as e: # Specifically catch the 'NoneType' object is not iterable error main_logger.warning(f"stripe-webhook warning: Email preparation issue, possibly due to timing: {e}") # Consider implementing a retry mechanism here if needed # For example, add the email task to a queue to try again later except Exception as e: main_logger.error(f"stripe-webhook error: Failed to send email: {e}") # Continue processing - we've already done inventory updates elif event["type"] == "checkout.session.async_payment_failed": main_logger.error( f"stripe-webhook error: Stripe payment error. {event['type']}" ) raise HTTPException(status_code=400, detail="Stripe payment error.") else: main_logger.error(f"stripe-webhook error: Unhandled event type {event['type']}") raise HTTPException(status_code=400, detail="Stripe payment error.") except Exception as e: main_logger.error(f"stripe-webhook error: Unhandled exception: {e}") import traceback main_logger.error(traceback.format_exc()) # Still return success to prevent Stripe from retrying the webhook # -- return 200 return {"success": True} @app.post("/create-now-payments", response_class=RedirectResponse, status_code=303) async def create_now_payments( request: Request, order_id: str = Form(...), weight: int = Form(...), email: EmailStr = Form(...), first_name: str | None = Form(None), last_name: str = Form(...), address: str = Form(...), address_no: str = Form(...), address_extra: str | None = Form(None), postal_code: str = Form(...), city: str = Form(...), country: str = Form(...), phone_number: str = Form(None), note: str | None = Form(None), ): """NOWPayments checkout session. User selects with which currency to pay. Sort of a custom NOW Payments Invoice page. """ # -- initial form-data validation form_data = await request.form() try: form = form_validation_checkout_session(form_data) except ValidationError as e: for error in e.errors(): main_logger.error(f"form-validation-stripe-checkout-session => {error}") # -- check if order-id is unique orders_settings = { "git_repo": settings.git_repo, "document_match": settings.document_match, "block_types": settings.block_types, } if not is_order_id_open(orders_settings, form.order_id): main_logger.error("now-payments-checkout error: Order ID is wrong.") raise HTTPException(status_code=422, detail="Order ID is wrong.") shipping_info = { "email": form.email, "first_name": form.first_name, "last_name": form.last_name, "address": form.address, "address_no": form.address_no, "address_extra": form.address_extra, "postal_code": form.postal_code, "city": form.city, "country": form.country, "phone_number": form.phone_number, "note": form.note, } shipping_info = ShippingInfo(**shipping_info) session_data = request.session initialize_cart(session_data, settings.currency.default) if "cart" not in session_data and len(session_data["cart"].keys()) == 0: main_logger.error("now-payments-checkout error: Cart is empty.") raise HTTPException(status_code=403, detail="Cart is empty.") cart = CartSession(**request.session["cart"]) data = { "price_amount": cart.meta.total, "price_currency": settings.currency.now_payments, "order_id": form.order_id, "ipn_callback_url": f"{request.base_url}now-webhook", "success_url": f"{request.base_url}checkout/success", "cancel_url": f"{request.base_url}checkout", } data = NowPaymentsInvoice(**data) payment_url, np_order_id = await generate_payment_link(data) product_list = create_product_list(settings, cart.items) # -- save transaction to local-db customer_order = prepare_customer_order_data( settings.git_repo, settings.document_match, settings.document_exclude, settings.block_types, np_order_id, product_list, "NOW Payments", data.order_id, cart, shipping_info, ) save_shipping_info(customer_order, settings.git_repo, settings.local_db.filepath) # -- redirect to NOWPayments Checkout page return payment_url @app.post("/now-webhook") async def now_webhook(request: Request, background_tasks: BackgroundTasks): """NOW Payments IPN (Instant Payment Notification) webhook. Send email after successful payment confirmation. """ main_logger.info("now-webhook: Starting processing") try: payload = await request.json() main_logger.info(f"now-webhook: Received payload with status: {payload.get('payment_status', 'unknown')}") sig_header = request.headers["x-nowpayments-sig"] if os.getenv("ENV") == "production": ipn_secret_key = os.getenv("NOWPAYMENTS_IPN") else: ipn_secret_key = os.getenv("NOWPAYMENTS_SANDBOX_IPN") is_verified = ipn_signature_check(ipn_secret_key, sig_header, payload) if is_verified: # check payment status, update order if # status has changed (finished, expired, failed) # send email if status: finished. event_type = payload["payment_status"] main_logger.info(f"now-webhook: Processing verified event type: {event_type}") if event_type in ["finished", "expired", "failed"]: if "payment_id" in payload: order_id = payload["order_id"] main_logger.info(f"now-webhook: Processing order: {order_id}") status_res = update_status_order( event_type, order_id, settings.git_repo, settings.local_db.filepath, ) if status_res is False: main_logger.error(f"now-webhook: Failed to update order status for {order_id}") return JSONResponse(content={"success": False, "error": "Order could not be updated."}, status_code=400) if event_type == "finished": product_list = fetch_order_product_list( order_id, settings.git_repo, settings.local_db.filepath ) main_logger.info(f"now-webhook: Found {len(product_list) if product_list else 0} products to update") if product_list: updated_products = 0 for product_info in product_list: try: product = read_file( settings.git_repo, f"products/{product_info.filename}", settings.document_match, settings.block_types, ) # Get the variation ID based on size and style product_variation_id = get_variation_id(product, product_info.size, product_info.style) main_logger.info(f"now-webhook: Updating inventory for {product_info.filename}, variation: {product_variation_id}, quantity: {product_info.quantity}") update_product_inventory( product_info.filename, product_info.quantity, product_variation_id, settings, ) updated_products += 1 except Exception as e: main_logger.error(f"now-webhook error: Failed to update inventory for {product_info.filename}: {e}") import traceback main_logger.error(traceback.format_exc()) # Continue processing other products main_logger.info(f"now-webhook: Successfully updated {updated_products} out of {len(product_list)} products") # -- send email try: email_settings = { "git_repo": settings.git_repo, "document_match": settings.document_match, "block_types": settings.block_types, } try: main_logger.info(f"now-webhook: Preparing email for order {order_id}") order = prepare_email_order(order_id, email_settings) main_logger.info(f"now-webhook: Sending email for order {order_id}") await send_email(email_settings, order, background_tasks) main_logger.info(f"now-webhook: Email queued successfully") except TypeError as e: main_logger.warning(f"now-webhook warning: Email preparation issue: {e}") import traceback main_logger.error(traceback.format_exc()) except Exception as e: main_logger.error(f"now-webhook error: Failed to send email: {e}") import traceback main_logger.error(traceback.format_exc()) elif event_type == "failed": main_logger.error(f"now-webhook: Payment failed for order {order_id}") # We don't want to raise an exception here since we need to return a success response # Just log the error else: main_logger.error("now-webhook: payment_id not in payload") return JSONResponse(content={"success": False, "error": "Payment ID missing"}, status_code=400) else: main_logger.info(f"now-webhook: Event type {event_type} not processed (not in finished/expired/failed)") else: main_logger.error("now-webhook: Signature verification failed") return JSONResponse(content={"success": False, "error": "Signature verification failed"}, status_code=400) except Exception as e: main_logger.error(f"now-webhook error: Unhandled exception: {e}") import traceback main_logger.error(traceback.format_exc()) # Return success anyway to prevent NOW Payments from retrying constantly main_logger.info("now-webhook: Processing completed") return JSONResponse(content={"success": True}, status_code=200) @app.get("/checkout/success", response_class=HTMLResponse) async def checkout_success(request: Request): """View that confirms to the user that the checkout procedure was successful.""" session_data = request.session # if there's no cart session, redirect the user to `/` if "cart" not in session_data: clear_cart(session_data, settings.currency.default) return RedirectResponse(f"{request.base_url}", status_code=303) cart = CartSession(**request.session["cart"]) filename = "orders" orders = read_file( settings.git_repo, filename, settings.document_match, settings.block_types ) customer_order = [ order for order in orders.blocks if order.order_id == cart.meta.order_id ] # if cart.meta.order_id does not match any order, redirect the user to `/` if len(customer_order) == 0: clear_cart(session_data, settings.currency.default) return RedirectResponse(f"{request.base_url}", status_code=303) # -- save data and clear cart, return view data = { "title": "Checkout Success", "order_id": cart.meta.order_id, "customer_email": customer_order[0].shipping_info.email, } # -- prepare order to xlsx format product_list = fetch_order_product_list( cart.meta.order_id, settings.git_repo, settings.local_db.filepath ) if not product_list: raise HTTPException(status_code=403, detail="Order checkout is missing information.") products_shipping_meta = [] for product_info in product_list: product = read_file(settings.git_repo, f"products/{product_info.filename}", settings.document_match, settings.block_types) products_shipping_meta.append( ProductShippingMeta( weight=product.meta.weight, customs_tariff_code=product.meta.customs_tariff_code, origin_country_iso=product.meta.origin_country_iso, ) ) email_from = os.getenv("MAIL_FROM") export_order_to_xlsx( customer_order[0], products_shipping_meta, settings.order_upload.filepath, settings.currency.default, settings.git_repo, email_from, ) clear_cart(session_data, settings.currency.default) cart = CartSession(**request.session["cart"]) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) return templates.TemplateResponse( request, "checkout-success.html", {"request": request, "site": site, "data": data, "cart": cart}, ) @app.get("/checkout/error", response_class=HTMLResponse) async def checkout_error(request: Request): """View for when the checkout procedure fails.""" cart = CartSession(**request.session["cart"]) site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) data = { "title": "Checkout", } return templates.TemplateResponse( request, "checkout-error.html", {"request": request, "site": site, "data": data, "cart": cart}, ) @app.post("/cart/clear") async def cart_clear( request: Request, js: bool = False, ): """Clear cart.""" request.session.clear() return RedirectResponse(request.base_url, status_code=303) # -- set custom HTTP errors @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): """Handle invalid data exception by returning an HTML response rather than a JSON one. """ site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) message = "We can't process your request." main_logger.error(exc) t = templates.TemplateResponse( request, "error.html", { "request": request, "site": site, "data": { "title": "Error", "error": message, "status_code": 422, }, "cart": None, }, ) return HTMLResponse(content=t.body, status_code=422) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request, exc): """Custom handling of HTTP exception (400-500). Prepare an appropriate user-facing message and pass it to the error.html template. """ message = "" main_logger.error(f"http-exception: error => {exc}") if exc.status_code == 404: message = "Nothing was found here." elif exc.status_code == 500: message = "Server error." elif 400 <= exc.status_code <= 499: message = "Something went wrong." if exc.detail: message = exc.detail site = read_file( settings.git_repo, "site", settings.document_match, settings.block_types, ) data = { "title": "Error", "error": message, "status_code": exc.status_code, } t = templates.TemplateResponse( request, "error.html", { "request": request, "site": site, "data": data, "cart": None, }, ) return HTMLResponse(content=t.body, status_code=exc.status_code)