ObjServiceFHShopify API

Method	Signature	Description
check_preflight	`check_preflight() -> bool`	Check if all required definitions are present.
get_colour_display_name	`get_colour_display_name(colour_code: str) -> str`	Map a raw colour code to a clean display
create_dryrun_table	`create_dryrun_table()`	Create the dry-run verification table if it doesn't exist.
compare_dryrun_results	`compare_dryrun_results(run1_start = None, run1_end = None, run2_start = None, run2_end = None, tolerance = 0.01)`	Compare two dry-run datasets to verify refactored calculations.
get_dryrun_summary	`get_dryrun_summary(start_timestamp = None, end_timestamp = None)`	Get summary statistics for a dry-run.
clear_dryrun_data	`clear_dryrun_data(before_timestamp = None)`	Clear dry-run data from the verification table.
rate_limit	`rate_limit()`
set_stage_id	`set_stage_id(stage_id)`
patch_param	`patch_param(Text)`
rate_limit_graphql	`rate_limit_graphql(cost_data)`	Handle GraphQL cost-based rate limiting
ReadConnection	`ReadConnection(Service)`
Read	`Read(Service, Param1 = '', Param2 = '', Param3 = '')`
Update	`Update(Message = '')`
Connect	`Connect()`
Send	`Send()`
create_body	`create_body(Param1, Promotion = '')`
bloom_reject	`bloom_reject(Sku, Reason = '')`
BloomProduct	`BloomProduct(P, InboundSku)`
BloomMeta	`BloomMeta(Param1 = '', Param2 = '', Sku = '', JsonPack = '')`
RetrieveSet	`RetrieveSet(Param1 = '', Param2 = '', Param3 = '')`	Retrieve product data using GraphQL API
retrieve_all_products	`retrieve_all_products(page_size = 50, max_products = None)`	Retrieve all products from Shopify using GraphQL with pagination.
get_promotions	`get_promotions(window_past_days: int = 30, window_future_days: int = 30) -> dict`	Retrieve past, current, and future promotions from
build_status_email	`build_status_email() -> str`	Build an HTML status email for the Shopify integration.
send_status_email	`send_status_email(to_address: str) -> str`	Build and send the Shopify status email.
recon_promotions	`recon_promotions() -> dict`	Reconcile Trader promotions against Shopify prices.
get_product_by_sku	`get_product_by_sku(sku: str) -> dict`	Retrieve a single product from Shopify by SKU.
get_product_set	`get_product_set(product_set: str) -> list`	Resolve a product set name to a list of SKUs.
get_group_skus	`get_group_skus(group_guid: str) -> list`	Resolve a variant group GUID to its
clean_previous_promo_tags	`clean_previous_promo_tags(on_progress = None)`	Remove 'special' and promo name tags from
push_products_batch	`push_products_batch(product_set: str = '', sku_list: list = None, max_workers: int = 3, on_progress = None) -> dict`	Push multiple products to Shopify.
list_products	`list_products(page_size: int = 50, max_products: int = None, save_to_db: bool = False) -> dict`	Retrieve all products from Shopify and return as
save_products_to_inventory	`save_products_to_inventory(products, sku = None, promo_data = None)`	Save retrieved products to data_shopify_inventory table.
is_product_cache_valid	`is_product_cache_valid(sku, max_age_hours = 24)`	Check if product data in inventory cache is still valid.
update_inventory_sync_time	`update_inventory_sync_time(sku)`	Update the last_sync timestamp for a product in inventory cache.
sync_product_status_from_shopify	`sync_product_status_from_shopify(product_id = None, sku = None, web_status_field = 'PM_WEBACTIVE')`	Sync product status from Shopify back to local database.
sync_all_product_statuses	`sync_all_product_statuses(web_status_field = 'PM_WEBACTIVE', max_products = None)`	Sync status for all products from Shopify to database.
DeleteProduct	`DeleteProduct(Param1 = '', Param2 = '', Param3 = '', DoTrace = 1)`	Delete product using GraphQL API
DeleteSku	`DeleteSku(Key)`
FoldSet	`FoldSet(Param1 = '', Param2 = '', Param3 = '')`	Remove duplicate products for a given SKU using GraphQL.
update_images	`update_images(base_sku, product_id, variant_images)`	Upload product images via GraphQL
create_product	`create_product(sku, guid = '', product_data = None, option_level = 1)`	Create or update a product using GraphQL Admin API.

¶ ObjServiceFHShopify (API Reference)

¶ class ObjServiceApi(ObjApi.ObjApi)

¶ get_package()

¶ check_package()

¶ main_callback()

¶ sync()

¶ locations()

¶ singleton(sku: str = typer.Option(..., '--sku', '-s', help='Comma-separated SKUs, OR a single GROUP- identifier — group is expanded into one product with N variants'))

¶ delete(sku: str = typer.Option(None, '--sku', '-s', help='Comma-separated SKU codes to delete'), product: str = typer.Option(None, '--product', '-p', help='Comma-separated Shopify product IDs to delete'))

¶ cache_config(set_max_age: int = typer.Option(None, '--set-max-age', '-s', help='Set cache max age in hours'), show: bool = typer.Option(True, '--show', help='Show current cache configuration'))

¶ promotions(past_days: int = typer.Option(30, '--past', '-p', help='Days back for ended promotions'), future_days: int = typer.Option(30, '--future', '-f', help='Days ahead for upcoming promotions'), format_output: str = typer.Option('summary', '--format', help='Output: summary, json'))

¶ recon(format_output: str = typer.Option('summary', '--format', '-f', help='Output: summary, detailed, json'))

¶ product(sku: str = typer.Argument(..., help='Product SKU to look up'), format_output: str = typer.Option('summary', '--format', '-f', help='Output: summary, json'))

¶ batch_push(product_set: str = typer.Argument('promo', help='promo, active, old, or comma-separated SKUs'))

¶ status_email(to: str = typer.Option(None, '--to', '-t', help='Send to email address'), output_file: str = typer.Option(None, '--output', '-o', help='Save HTML to file'))

¶ calibrate(count: int = typer.Option(5, help='Number of products to push'), product_set: str = typer.Argument('active', help='Product set to calibrate with'))

¶ sync_audit(fix: bool = typer.Option(False, '--fix', '-f', help='Auto-fix mismatched products'))

¶ promo_email(to: str = typer.Option('', help='Email recipients (comma-separated). Defaults to shopify.notify_emails config.'))

¶ ObjServiceFHShopify (API Reference)

¶ class ObjServiceApi(ObjApi.ObjApi)

¶ get_package()

¶ check_package()

¶ main_callback()

¶ sync()

¶ update(product_set: str = typer.Argument(..., help='Product set: promo, new, old, active, variants, updates, all'))

¶ locations()

¶ singleton(sku: str = typer.Option(..., '--sku', '-s', help='Comma-separated SKUs, OR a single GROUP- identifier — group is expanded into one product with N variants'))

¶ delete(sku: str = typer.Option(None, '--sku', '-s', help='Comma-separated SKU codes to delete'), product: str = typer.Option(None, '--product', '-p', help='Comma-separated Shopify product IDs to delete'))

¶ cache_config(set_max_age: int = typer.Option(None, '--set-max-age', '-s', help='Set cache max age in hours'), show: bool = typer.Option(True, '--show', help='Show current cache configuration'))

¶ promotions(past_days: int = typer.Option(30, '--past', '-p', help='Days back for ended promotions'), future_days: int = typer.Option(30, '--future', '-f', help='Days ahead for upcoming promotions'), format_output: str = typer.Option('summary', '--format', help='Output: summary, json'))

¶ recon(format_output: str = typer.Option('summary', '--format', '-f', help='Output: summary, detailed, json'))

¶ product(sku: str = typer.Argument(..., help='Product SKU to look up'), format_output: str = typer.Option('summary', '--format', '-f', help='Output: summary, json'))

¶ batch_push(product_set: str = typer.Argument('promo', help='promo, active, old, or comma-separated SKUs'))

¶ status_email(to: str = typer.Option(None, '--to', '-t', help='Send to email address'), output_file: str = typer.Option(None, '--output', '-o', help='Save HTML to file'))

¶ calibrate(count: int = typer.Option(5, help='Number of products to push'), product_set: str = typer.Argument('active', help='Product set to calibrate with'))

¶ sync_audit(fix: bool = typer.Option(False, '--fix', '-f', help='Auto-fix mismatched products'))

¶ promo_email(to: str = typer.Option('', help='Email recipients (comma-separated). Defaults to shopify.notify_emails config.'))