Synchronise Stripe products

stripe_sync_products.py

import stripe

async def sync_products(
    live_key: str, sandbox_key: str, idempotency_suffix: str = "v0"
):
    products = await stripe.Product.list_async(api_key=live_key)

    for product in products:
        new_product = await stripe.Product.create_async(
            api_key=sandbox_key,
            name=product["name"],
            description=product.get("description"),
            active=product["active"],
            metadata=product.get("metadata", {}),
            tax_code=product["tax_code"],
            idempotency_key=product["id"] + idempotency_suffix,
        )

        await stripe.Product.modify_async(
            new_product.id,
            api_key=sandbox_key,
            idempotency_key=new_product.id + idempotency_suffix,
            metadata={"source_id": product["id"]},
        )

        prices = await stripe.Price.list_async(
            api_key=live_key,
            product=product["id"],
            expand=["data.tiers", "data.currency_options"],
        )

        async for price in prices.auto_paging_iter():
            tiers = price.get("tiers", [])
            tiered_pricing = []

            for tier in tiers:
                tiered_pricing.append(
                    {
                        "up_to": "inf"
                        if tier.get("up_to") is None
                        else tier.get("up_to"),
                        "unit_amount": tier["unit_amount"],
                        "flat_amount": tier["flat_amount"],
                    }
                )

            currency_options = {}
            for name, value in price.get("currency_options", {}).items():
                if name != price["currency"]:
                    currency_options[name] = {
                        "custom_unit_amount": value["custom_unit_amount"],
                        "tax_behavior": value["tax_behavior"],
                        "unit_amount": value["unit_amount"],
                    }

            new_price = await stripe.Price.create_async(
                api_key=sandbox_key,
                product=new_product["id"],
                unit_amount=price["unit_amount"],
                currency=price["currency"],
                currency_options=currency_options,
                recurring=price.get("recurring", {}),
                metadata=price.get("metadata", {}),
                idempotency_key=price["id"] + idempotency_suffix,
                billing_scheme=price["billing_scheme"],
                tiers_mode=price["tiers_mode"],
                tax_behavior=price["tax_behavior"],
                tiers=tiered_pricing if tiers else None,
                lookup_key=price.get("lookup_key"),
            )

            await stripe.Price.modify_async(
                new_price.id,
                api_key=sandbox_key,
                metadata={"source_id": price["id"]},
                idempotency_key=new_price["id"] + idempotency_suffix,
            )