Copy Stripe production products and prices to test environment

copy-stripe-setup.py

### The best way to start is to use the "Delete all test data" button at https://dashboard.stripe.com/test/developers

### Setup:

import stripe

test = 'sk_test_51IXXXyoursecretkey'

prod = 'sk_live_51IXXXyoursecretkey'

skip_fields = ['amount_decimal', 'unit_amount_decimal', 'type', 'object', 'created', 'livemode', 'updated',]
ignore_products = ['prod_JNXXXyourproductID',]

### Fetch production products and prices

stripe.api_key = prod
prodproducts = ('Product', stripe.Product.list(active=True))
prodprices = ('Price', stripe.Price.list(active=True))
stripe.api_key = test

### Define some functions

def clear_stripe_things(thing: tuple) -> None:
	'''
	Clear out test products/prices
	'''
	print('Clearing', thing[0])
	for p in getattr(stripe, thing[0]).list():
		print('Clearing', p.get('id'), '-', p.get('name'))
		if p.get('product') in ignore_products:
			print('Ignoring')
			continue
		if not p.get('active'):
			print('Inactive, skipping')
			continue
		if p.get('livemode'):
			print('LIVE MODE THING! Skipping')
			continue
		try:
			getattr(stripe, thing[0]).modify(p.get('id'), active=False)
		except Exception as e:
			print('Exception modifying for', p)
			print(e)
			pass
		try:
			p.delete()
		except Exception as e:
			print('Exception deleting for', p.get('id'))
			print(e)
			continue


def upload_products() -> None:
	'''
	Copy production products to test, preserving IDs
	'''
	print('Uploading products')
	up = list()
	for p in prodproducts[1].get('data'):
		print('Queueing', p.get('id'), '-', p.get('name'))
		if p.get('product') in ignore_products:
			print('Ignoring')
			continue
		up.append({k:v for k,v in p.items() if k not in skip_fields})
	for p in up:	
		try:
			print('Uploading', p.get('id'), '-', p.get('name'))
			print(up)
			stripe.Product.create(**p)
		except Exception as e:
			print('EXCEPTION creating', p)
			print('EXCEPTION:', e)
			continue


def get_by_kv(lis, k, v):
	'''
	Return first element in list where key, value matches k, v in dictionary
	'''
	for e in lis:
		if e.get(k) == v:
			return e
	

def upload_prices() -> None:
	'''
	Upload prices to Stripe, preserving product ID correspondences
	'''
	skips = skip_fields + ['id']
	for p in prodproducts[1]:
		print('Uploading for', p.get('id'), '-', p.get('name'))
		if not p.get('active'):
			print('Inactive product, skipping')
			continue
		# x =	get_by_kv(prodprices[1], 'product', p.get('id'))
		# print(x)		
		prod_price = get_by_kv(prodprices[1], 'product', p.get('id'))
		print('prod_price', prod_price)
		test_price = {k:v for k,v in prod_price.items() if k not in skips}
		print('test_price', test_price)
		try:
			stripe.Price.create(**test_price)
		except Exception as e:
			print('EXCEPTION creating price', p)
			print('EXCEPTION:', e)
			continue


stripe.api_key = test
clear_stripe_things(prodprices)
clear_stripe_things(prodproducts)


stripe.api_key=test
upload_products()
upload_prices()

not working...keep getting Exception messages. I'm using latest API level

Sorry about that. I'm not developing for Stripe anymore, unfortunately.

No worries, left a comment so future readers would know that it won't work with latest stripe API version

Thanks for this @mikegogulski . We're using tiered (graduated) pricing, and with some minor tweaks and the old stripe API version (v2.70.0), it happily copied the products and pricing over. Changes:

• Added an expand=['data.tiers'] parameter to the call to stripe.Price.list(...)
• Added unit_amount_decimal and flat_amount_decimal to the skip_fields array for the prices.
• Deleted the (nested) unit_amount_decimal and flat_amount_decimal from each item in the tiers array of a price.
• Set the up_to property of a tier to 'inf' if it isn't set.

pip install stripe==2.70.0

With the updates described above:

### The best way to start is to use the "Delete all test data" button at https://dashboard.stripe.com/test/developers

### Setup:

import stripe

test = 'sk_test_51IXXXyoursecretkey'

prod = 'sk_live_51IXXXyoursecretkey'

skip_fields = ['amount_decimal', 'unit_amount_decimal', 'type', 'object', 'created', 'livemode', 'updated',]
ignore_products = ['prod_JNXXXyourproductID',]

### Fetch production products and prices

stripe.api_key = prod
prodproducts = ('Product', stripe.Product.list(active=True))
prodprices = ('Price', stripe.Price.list(active=True, expand=['data.tiers']))
stripe.api_key = test

### Define some functions

def clear_stripe_things(thing: tuple) -> None:
  '''
  Clear out test products/prices
  '''
  print('Clearing', thing[0])
  for p in getattr(stripe, thing[0]).list():
    print('Clearing', p.get('id'), '-', p.get('name'))
    if p.get('product') in ignore_products:
      print('Ignoring')
      continue
    if not p.get('active'):
      print('Inactive, skipping')
      continue
    if p.get('livemode'):
      print('LIVE MODE THING! Skipping')
      continue
    try:
      getattr(stripe, thing[0]).modify(p.get('id'), active=False)
    except Exception as e:
      print('Exception modifying for', p)
      print(e)
      pass
    try:
      p.delete()
    except Exception as e:
      print('Exception deleting for', p.get('id'))
      print(e)
      continue


def upload_products() -> None:
  '''
  Copy production products to test, preserving IDs
  '''
  print('Uploading products')
  up = list()
  for p in prodproducts[1].get('data'):
    print('Queueing', p.get('id'), '-', p.get('name'))
    if p.get('product') in ignore_products:
      print('Ignoring')
      continue
    up.append({k:v for k,v in p.items() if k not in skip_fields})
  for p in up:	
    try:
      del p['default_price']
      print('Uploading', p.get('id'), '-', p.get('name'))
      print(up)
      stripe.Product.create(**p)
    except Exception as e:
      print('EXCEPTION creating', p)
      print('EXCEPTION:', e)
      continue


def get_by_kv(lis, k, v):
  '''
  Return first element in list where key, value matches k, v in dictionary
  '''
  for e in lis:
    if e.get(k) == v:
      return e
  

def upload_prices() -> None:
  '''
  Upload prices to Stripe, preserving product ID correspondences
  '''
  skips = skip_fields + ['id','unit_amount_decimal','flat_amount_decimal']
  for p in prodproducts[1]:
    print('Uploading for', p.get('id'), '-', p.get('name'))
    if not p.get('active'):
      print('Inactive product, skipping')
      continue
    # x =	get_by_kv(prodprices[1], 'product', p.get('id'))
    # print(x)		
    prod_price = get_by_kv(prodprices[1], 'product', p.get('id'))
    # remove the "flat_amount_decimal" and "unit_amount_decimal" keys from the tiers
    #
    for tier in prod_price.get('tiers'):
      del tier['flat_amount_decimal']
      del tier['unit_amount_decimal']
      tier['up_to'] = 'inf' if tier.get('up_to') == None else tier.get('up_to')
    print('prod_price', prod_price)
    test_price = {k:v for k,v in prod_price.items() if k not in skips}
    print('test_price', test_price)
    try:
      stripe.Price.create(**test_price)
    except Exception as e:
      print('EXCEPTION creating price', p)
      print('EXCEPTION:', e)
      continue


stripe.api_key = test
clear_stripe_things(prodprices)
clear_stripe_things(prodproducts)


stripe.api_key=test
upload_products()
upload_prices()