diff --git a/db/schema.sql b/db/schema.sql index 8d5a950..b0986ad 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -21,6 +21,8 @@ CREATE TABLE orders ( status order_status, -- References to the same order in Stripe and Printful. + -- `stripe_id` has an implicit unique index for efficient + -- queries. stripe_id TEXT UNIQUE NOT NULL, printful_id TEXT, @@ -60,5 +62,6 @@ CREATE TABLE order_items ( CREATE TABLE stripe_checkouts ( -- The Stripe checkout session used for idempotency protection id TEXT PRIMARY KEY, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + is_processed BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); \ No newline at end of file diff --git a/src/handlers/checkout.py b/src/handlers/checkout.py index 350fe02..9b0dbba 100644 --- a/src/handlers/checkout.py +++ b/src/handlers/checkout.py @@ -10,12 +10,11 @@ from lib.types import StripeCheckout - db = Database(url=os.getenv("DATABASE_URL")) stripe.api_key = get_api_key() -def begin_fulfillment(session_id: str, event_type: str): +def begin_fulfillment(session_id: str): """ Begin fulfilling a checkout session on successful payment, using an idempotent operation as required by Stripe. @@ -46,4 +45,4 @@ def process_webhook_request(payload: Any, signature: Any): event.type == "checkout.session.completed" or event.type == "checkout.session.async_payment_succeeded" ): - begin_fulfillment(event.data.object["id"], event.type) + begin_fulfillment(event.data.object["id"]) diff --git a/src/handlers/fulfillment.py b/src/handlers/fulfillment.py index d19597c..cc65283 100644 --- a/src/handlers/fulfillment.py +++ b/src/handlers/fulfillment.py @@ -1,8 +1,21 @@ +import os + +import requests import stripe from lib.stripe import get_api_key -from lib.types import StripeCheckout - +from lib.types import Order, OrderStatus, StripeCheckout +from lib.db import Database +from lib.printful import ( + PRINTFUL_STATUS_MAP, + PrintfulClient, + PrintfulItem, + PrintfulRecipient, +) +from lib.logs import Logs +db = Database(url=os.getenv("DATABASE_URL")) +printful = PrintfulClient(api_key=os.getenv("PRINTFUL_API_KEY")) +log = Logs(log_group=os.environ["LOG_GROUP"]) stripe.api_key = get_api_key() @@ -10,17 +23,64 @@ def process_fulfillment(checkout: StripeCheckout): """ Fulfills a successful checkout session. """ - # TODO: Make sure fulfillment hasn't already been - # performed for this checkout. - + # Fetch the items purchased and the receipt link. session = stripe.checkout.Session.retrieve( checkout.id, - expand=["line_items"], + expand=["line_items", "payment_intent.latest_charge"], ) if session.payment_status == "unpaid": return - # TODO: Perform fulfillment of the line items - # TODO: Record/save fulfillment status for this - pass + log.info(session.line_items.data) + order = Order( + email=session.customer_details.email, + stripe_id=checkout.id, + receipt_url=session.payment_intent.latest_charge.receipt_url, + price=session.amount_total, + items=[ + # TODO: Think about the items a little more. + ], + ) + + order, created = db.create_order(order) + if not created and order.printful_id is not None: + # Order was already fully fulfilled on a previous attempt. + return + + shipping = session.shipping_details + recipient = PrintfulRecipient( + name=shipping.name, + address1=shipping.address.line1, + address2=shipping.address.line2, + city=shipping.address.city, + state_code=shipping.address.state, + country_code=shipping.address.country, + zip=shipping.address.postal_code, + email=session.customer_details.email, + ) + items = [ + PrintfulItem(product_id=int(item.product_id), quantity=item.quantity) + for item in order.items + ] + + # Perform a lookup so we avoid creating duplicate errors if we already processed + # this in a previous call. + try: + result = printful.get_order_by_external_id(order.id) + except requests.HTTPError as e: + if e.response.status_code == 404: + result = printful.create_order( + recipient=recipient, + items=items, + external_id=checkout.id, + ) + else: + raise e + + printful_order = result["result"] + db.update_order( + order.id, + printful_id=str(printful_order["id"]), + status=PRINTFUL_STATUS_MAP.get(printful_order["status"], OrderStatus.pending), + ) diff --git a/src/handlers/printful.py b/src/handlers/printful.py index 3dedf15..99e39a8 100644 --- a/src/handlers/printful.py +++ b/src/handlers/printful.py @@ -15,17 +15,6 @@ notify = Notify(topic_arn=os.environ["SNS_TOPIC_ARN"], phone=os.environ["NOTIFY_PHONE"]) queue = Queue(queue_url=os.environ["PRINTFUL_QUEUE_URL"]) -# Capture the states of an order from Printful, and for some of these, such as "onhold," -# we prefer to deal with them in the application-level, rather than persist such states -# to the database. -PRINTFUL_STATUS_MAP = { - "pending": OrderStatus.pending, - "inprocess": OrderStatus.pending, - "onhold": OrderStatus.pending, - "inreview": OrderStatus.pending, - "fulfilled": OrderStatus.fulfilled, - "failed": OrderStatus.failed, -} SUPPORTED_EVENTS = { "package_shipped", diff --git a/src/lib/db.py b/src/lib/db.py index f31f63c..0678fbc 100644 --- a/src/lib/db.py +++ b/src/lib/db.py @@ -1,7 +1,7 @@ from datetime import datetime from typing import Optional -from sqlalchemy import create_engine +from sqlalchemy import create_engine, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -45,15 +45,24 @@ def upsert_shipment(self, shipment: Shipment) -> Shipment: session.refresh(shipment) return shipment - def create_order(self, order: Order) -> Order: + def create_order(self, order: Order) -> tuple[Order, bool]: """ Write an order to the database, including its items. + Returns the order and whether it was newly created. """ with Session(self.engine) as session: - session.add(order) - session.commit() - session.refresh(order) - return order + try: + session.add(order) + session.commit() + session.refresh(order) + return order, True + except IntegrityError: + session.rollback() + existing = session.execute( + select(Order).where(Order.stripe_id == order.stripe_id) + ).scalar_one() + + return existing, False def record_stripe_checkout(self, checkout: StripeCheckout) -> bool: """ diff --git a/src/lib/printful.py b/src/lib/printful.py index 61ec328..a14cd35 100644 --- a/src/lib/printful.py +++ b/src/lib/printful.py @@ -2,6 +2,20 @@ from typing import Optional import requests +from lib.types import OrderStatus + +# Capture the states of an order from Printful, and for some of these, such as "onhold," +# we prefer to deal with them in the application-level, rather than persist such states +# to the database. +PRINTFUL_STATUS_MAP = { + "pending": OrderStatus.pending, + "inprocess": OrderStatus.pending, + "onhold": OrderStatus.pending, + "inreview": OrderStatus.pending, + "fulfilled": OrderStatus.fulfilled, + "failed": OrderStatus.failed, +} + @dataclass class PrintfulItem: @@ -70,3 +84,9 @@ def get_order(self, order_id: int): Send a request to get the order with `order_id`. """ return self._request("GET", f"/orders/{order_id}") + + def get_order_by_external_id(self, external_id: str): + """ + Send a request to get the order with `external_id`. + """ + return self._request("GET", f"/orders/@{external_id}") diff --git a/src/lib/types.py b/src/lib/types.py index 12b0e33..d996454 100644 --- a/src/lib/types.py +++ b/src/lib/types.py @@ -61,4 +61,5 @@ class StripeCheckout(Base): __tablename__ = "stripe_checkouts" id: Mapped[str] = mapped_column(primary_key=True) + is_processed: Mapped[bool] = mapped_column(default=False) created_at: Mapped[datetime] = mapped_column(default=datetime.now)