Shopify API

Shopify API Integration

Implements production-grade Shopify API integration for Admin REST API, Admin GraphQL API, and Storefront API. When loaded, this skill makes the model implement the shopifyapi Python SDK patterns including: OAuth 2.0 app installation flow (authorization_code grant), Admin REST API with cursor-based pagination (page_info, limit), Admin GraphQL API with rate limit handling, Storefront API for customer-facing experiences, webhook HMAC-SHA256 signature verification, API versioning (2024-01, etc.), and private app vs custom app authentication patterns.

TL;DR Checklist

• Use shopifyapi Python SDK (pip install ShopifyAPI)
• OAuth flow: redirect to /admin/oauth/authorize → POST /admin/oauth/access_token
• API versioning: use YYYY-MM format in api_version (e.g., "2024-01")
• Webhook verification: HMAC-SHA256 of request body with API_SECRET_KEY
• Cursor pagination: page_info param; use Link header for prev/next
• Private apps: use API_KEY + PASSWORD (deprecated; prefer custom apps)
• Custom apps: use access token from OAuth flow
• Storefront API: uses separate StorefrontAccessToken, not Admin token
• Environment variables: SHOPIFY_API_KEY, SHOPIFY_API_SECRET, SHOPIFY_API_VERSION, SHOPIFY_SCOPES

Core Workflow

1. Configure Shopify App: Set up app in Shopify Partners: App URL, Allowed redirection URL(s), Scopes (read_products, write_orders, etc.). Checkpoint: Note API_KEY (Client ID) and API_SECRET_KEY (Client Secret).
2. Initiate OAuth Flow: Redirect merchant to: https://{shop}.myshopify.com/admin/oauth/authorize?client_id={api_key}&scope={scopes}&redirect_uri={redirect_uri}&state={nonce}. Checkpoint: Generate and store state parameter to prevent CSRF.
3. Exchange Authorization Code: Merchant approves, Shopify redirects to your redirect_uri with code and hmac and shop and state. Verify state matches. Call POST https://{shop}.myshopify.com/admin/oauth/access_token with client_id, client_secret, code. Checkpoint: Response gives access_token (permanent, store securely).
4. Initialize Shopify Client: Use ShopifyAPI.Session or shopify.Session with shop, api_version, token. Activate session for SDK calls. Checkpoint: Verify by calling Shopify.Shop.current().
5. Call APIs: Use: (a) Admin REST: Shopify.Product.find(), Shopify.Order.find(), etc. with cursor pagination; (b) Admin GraphQL: Shopify.GraphQL(); (c) Storefront: Separate token via StorefrontAccessToken API. Checkpoint: Handle 429 Too Many Requests with Retry-After header.
6. Verify Webhooks: Compute HMAC-SHA256 of request body using API_SECRET_KEY (base64-decode key first). Compare with X-Shopify-Hmac-Sha256 header. Checkpoint: Verify before processing. Return 200 OK after.

Implementation Patterns

Pattern 1: OAuth 2.0 Flow and Client Initialization

"""Shopify OAuth 2.0 and client initialization patterns.

SDK: pip install ShopifyAPI (note: capital S in PyPI package name)

Shopify has two authentication modes:
1. OAuth 2.0 (public/custom apps) — RECOMMENDED
2. Private app authentication (API_KEY + PASSWORD) — DEPRECATED

App types:
- Public app: Installable by any merchant, listed in App Store
- Custom app: Built for one specific merchant, not in App Store
- Private app: Legacy, token per merchant (deprecated; use custom apps)
"""

from __future__ import annotations

import os
import hmac
import hashlib
import base64
import secrets
import logging
from typing import Any, Optional
from urllib.parse import urlencode

logger = logging.getLogger(__name__)


# ===================================================================
# ❌ BAD — hardcoded credentials, no state (CSRF), no validation
# ===================================================================

def bad_oauth_redirect_bad(shop: str) -> str:
    """❌ BAD: No state parameter (CSRF risk), hardcoded scopes."
    api_key = "d0...hardcoded_key_here"  # ❌ HARDCODED!
    
    # ❌ No state parameter = CSRF vulnerability
    return (
        f"https://{shop}/admin/oauth/authorize?"
        f"client_id={api_key}&"
        f"scope=read_products,write_orders&"  # ❌ Hardcoded
        f"redirect_uri=https://example.com/callback"  # ❌ Hardcoded
    )


# ===================================================================
# ✅ GOOD — environment-based, state for CSRF, proper validation
# ===================================================================

try:
    import shopify
    SHOPIFY_SDK_AVAILABLE = True
except ImportError:
    # Package may also be imported as ShopifyAPI
    try:
        import ShopifyAPI as shopify
        SHOPIFY_SDK_AVAILABLE = True
    except ImportError:
        SHOPIFY_SDK_AVAILABLE = False


class ShopifyConfig:
    """Typed configuration for Shopify app."""
    
    def __init__(
        self,
        api_key: str | None = None,
        api_secret: str | None = None,
        api_version: str | None = None,
        scopes: str | None = None,
        redirect_uri: str | None = None,
        app_url: str | None = None,
    ):
        self.api_key = api_key or os.environ.get("SHOPIFY_API_KEY", "")
        self.api_secret = api_secret or os.environ.get("SHOPIFY_API_SECRET", "")
        self.api_version = api_version or os.environ.get("SHOPIFY_API_VERSION", "2024-01")
        self.scopes = scopes or os.environ.get("SHOPIFY_SCOPES", "read_products")
        self.redirect_uri = redirect_uri or os.environ.get("SHOPIFY_REDIRECT_URI", "")
        self.app_url = app_url or os.environ.get("SHOPIFY_APP_URL", "")
    
    def validate(self) -> None:
        if not self.api_key:
            raise ValueError("SHOPIFY_API_KEY not configured")
        if not self.api_secret:
            raise ValueError("SHOPIFY_API_SECRET not configured")


def get_shopify_config() -> ShopifyConfig:
    config = ShopifyConfig()
    config.validate()
    return config


def generate_oauth_state() -> str:
    """Generate a secure random state parameter for CSRF protection.

    Store this value in session/cookie before redirecting to Shopify,
    then verify it matches the state parameter in the callback.
    """
    return secrets.token_urlsafe(32)


def sanitize_shop_domain(shop: str) -> str:
    """Sanitize and validate the shop domain parameter.

    Shopify shop domains must end with .myshopify.com.
    This prevents open redirect attacks.

    Args:
        shop: The shop parameter from request (e.g., "example.myshopify.com")

    Returns:
        Sanitized shop domain.

    Raises:
        ValueError: If invalid shop domain format.
    """
    if not shop:
        raise ValueError("Shop domain is required")
    
    # Remove https://, http:// prefixes if present
    shop = shop.lower().strip()
    if shop.startswith("https://"):
        shop = shop[8:]
    if shop.startswith("http://"):
        shop = shop[7:]
    
    # Remove trailing slash
    shop = shop.rstrip("/")
    
    # Must end with .myshopify.com
    if not shop.endswith(".myshopify.com"):
        raise ValueError(f"Invalid shop domain: {shop}")
    
    # Basic sanity check for allowed characters
    allowed_chars = set("abcdefghijklmnopqrstuvwxyz0123456789-.")
    if any(c not in allowed_chars for c in shop):
        raise ValueError(f"Invalid characters in shop domain: {shop}")
    
    return shop


def build_oauth_redirect_url(
    shop: str,
    state: str,
    config: ShopifyConfig | None = None,
) -> str:
    """Build the Shopify OAuth authorization URL.

    This is where you redirect the merchant to install your app.

    Args:
        shop: Sanitized .myshopify.com domain
        state: CSRF protection token (store this in session)
        config: Optional ShopifyConfig

    Returns:
        Full authorization URL to redirect to.
    """
    actual_config = config or get_shopify_config()
    sanitized_shop = sanitize_shop_domain(shop)
    
    params = {
        "client_id": actual_config.api_key,
        "scope": actual_config.scopes,
        "redirect_uri": actual_config.redirect_uri,
        "state": state,
        "grant_options[]": "",  # "per-user" for online access mode, empty for offline
    }
    
    query_string = urlencode(params)
    return f"https://{sanitized_shop}/admin/oauth/authorize?{query_string}"


def exchange_authorization_code(
    shop: str,
    code: str,
    config: ShopifyConfig | None = None,
) -> dict[str, Any]:
    """Exchange authorization code for access token (offline access mode).

    Shopify redirects to your redirect_uri with:
    - code: Authorization code (short-lived, 1-time use)
    - hmac: For verification
    - shop: The .myshopify.com domain
    - state: Your CSRF token

    Call this AFTER:
    1. Verifying state matches your stored state
    2. (Optional but recommended) Verifying the hmac parameter

    Args:
        shop: The .myshopify.com domain
        code: The authorization code from callback
        config: Optional config

    Returns:
        Dict with:
        - access_token: Permanent token (store securely!)
        - scope: Granted scopes
        - associated_user: (only for online access mode)
    """
    import requests
    
    actual_config = config or get_shopify_config()
    sanitized_shop = sanitize_shop_domain(shop)
    
    url = f"https://{sanitized_shop}/admin/oauth/access_token"
    
    body = {
        "client_id": actual_config.api_key,
        "client_secret": actual_config.api_secret,
        "code": code,
    }
    
    response = requests.post(url, json=body, timeout=30)
    
    if response.status_code != 200:
        logger.error(
            "Shopify OAuth token exchange failed: status=%d body=%s",
            response.status_code, response.text
        )
        raise RuntimeError(f"Failed to exchange code: {response.status_code}")
    
    data = response.json()
    
    return {
        "access_token": data.get("access_token"),
        "scope": data.get("scope"),
        "associated_user": data.get("associated_user"),
        "expires_in": data.get("expires_in"),  # Only for online tokens
        "associated_user_scope": data.get("associated_user_scope"),
    }


def verify_oauth_callback_hmac(
    params: dict[str, str],  # All query params from callback URL
    config: ShopifyConfig | None = None,
) -> bool:
    """Verify the HMAC signature on OAuth callback parameters.

    Shopify signs the callback query params (code, shop, state, timestamp)
    with your API secret.

    Args:
        params: Dictionary of query params (code, hmac, shop, state, timestamp)
        config: Optional config

    Returns:
        True if signature valid.

    Raises:
        ValueError: If verification fails or no hmac present.
    """
    actual_config = config or get_shopify_config()
    
    received_hmac = params.get("hmac", "")
    if not received_hmac:
        raise ValueError("No hmac parameter in callback")
    
    # Build message: sorted key=value pairs joined by &, excluding hmac
    # Keys: code, shop, state, timestamp (alphabetical order EXCEPT hmac)
    sorted_params = sorted(params.items())
    message_parts = []
    for key, value in sorted_params:
        if key != "hmac":
            message_parts.append(f"{key}={value}")
    
    message = "&".join(message_parts)
    
    # Compute HMAC-SHA256
    key_bytes = actual_config.api_secret.encode("utf-8")
    message_bytes = message.encode("utf-8")
    
    computed_hmac = hmac.new(key_bytes, message_bytes, hashlib.sha256).hexdigest()
    
    # Constant-time comparison
    if hmac.compare_digest(computed_hmac, received_hmac):
        logger.info("Shopify OAuth callback HMAC verified")
        return True
    else:
        logger.warning(
            "Shopify OAuth HMAC mismatch: computed=%s received=%s",
            computed_hmac, received_hmac
        )
        raise ValueError("OAuth callback HMAC verification failed")


def init_shopify_session(
    shop: str,
    access_token: str,
    config: ShopifyConfig | None = None,
) -> Any:
    """Initialize and activate a Shopify API session.

    After this call, you can use:
    - shopify.Product.find()
    - shopify.Order.find()
    - shopify.GraphQL().execute()
    - etc.

    Args:
        shop: .myshopify.com domain
        access_token: OAuth access token
        config: Optional config

    Returns:
        The activated session.
    """
    if not SHOPIFY_SDK_AVAILABLE:
        raise RuntimeError("Shopify SDK not installed. pip install ShopifyAPI")
    
    actual_config = config or get_shopify_config()
    sanitized_shop = sanitize_shop_domain(shop)
    
    # Clear any previous session
    shopify.ShopifyResource.clear_session()
    
    # Create new session
    session = shopify.Session(
        sanitized_shop,
        actual_config.api_version,
        access_token,
    )
    
    # Activate it
    shopify.ShopifyResource.activate_session(session)
    
    logger.info("Shopify session activated for shop: %s", sanitized_shop)
    return session


def verify_access_token(shop: str, access_token: str) -> dict[str, Any]:
    """Verify that an access token is still valid by fetching Shop.current().

    Call this:
    - After OAuth to verify token works
    - Periodically to check if app was uninstalled (token revoked)

    Returns:
        Shop info dict.
    """
    init_shopify_session(shop, access_token)
    
    try:
        shop_info = shopify.Shop.current()
        return {
            "id": shop_info.id,
            "name": shop_info.name,
            "email": shop_info.email,
            "domain": shop_info.domain,
            "myshopify_domain": shop_info.myshopify_domain,
            "currency": shop_info.currency,
            "timezone": shop_info.timezone,
        }
    except Exception as e:
        # Common errors:
        # - 401 Unauthorized: Token invalid/revoked (app uninstalled)
        # - 403 Forbidden: Token doesn't have required scope
        logger.error("Shopify access token verification failed: %s", e)
        raise ValueError(f"Access token invalid or revoked: {e}") from e

### Pattern 2: Admin REST API with Cursor Pagination
```python
... (additional code patterns as necessary)...

Pattern 2: REST API with Rate Limiting and Retries

import logging
import time
from typing import Optional


logger = logging.getLogger(__name__)


class ShopifyRESTClient:
    """Shopify REST API client with rate limit handling."""

    def __init__(self, shop_domain: str, access_token: str):
        self._shop = shop_domain
        self._token = access_token

    def get_order(self, order_id: int) -> dict:
        """Fetch a single order by ID."""
        url = f"https://{self._shop}.myshopify.com/admin/api/2024-01/orders/{order_id}.json"
        headers = {"X-Shopify-Access-Token": self._token}
        # In production: requests.get(url, headers=headers)
        return {}

    def list_orders(self, status: str = "any", limit: int = 50) -> list[dict]:
        """List orders with filtering and pagination."""
        params = {"status": status, "limit": limit}
        url = f"https://{self._shop}.myshopify.com/admin/api/2024-01/orders.json"
        headers = {"X-Shopify-Access-Token": self._token}
        # In production: requests.get(url, params=params, headers=headers)
        return []

    def update_inventory(self, variant_id: int, quantity: int) -> dict:
        """Update inventory level for a product variant."""
        body = {"inventory_item_id": variant_id, "available": quantity}
        url = f"https://{self._shop}.myshopify.com/admin/api/2024-01/inventory_levels/adjust_quantity.json"
        headers = {"X-Shopify-Access-Token": self._token, "Content-Type": "application/json"}
        # In production: requests.put(url, json=body, headers=headers)
        return {}


class RateLimitState:
    """Tracks API rate limit status."""

    def __init__(self):
        self.remaining = 40
        self.reset_time = time.time()

    def is_rate_limited(self) -> bool:
        return time.time() > self.reset_time or self.remaining <= 0

    def update_from_headers(self, headers: dict) -> None:
        if "X-Shopify-Shop-Api-Call-Limit" in headers:
            parts = headers["X-Shopify-Shop-Api-Call-Limit"].split("/")
            self.remaining = int(parts[0])

Constraints

MUST DO

• Implement structured error responses with consistent format: {error_code, message, details, request_id}
• Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm
• Validate all incoming requests against a schema before processing — reject malformed input with clear error messages
• Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries

MUST NOT DO

• Do not expose internal implementation details, stack traces, or database queries in error responses
• Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits
• Never trust client-supplied authentication tokens without validation (signature verification, expiration check)
• Do not log request/response bodies containing PII, API keys, or other sensitive data

Live References

Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

• Shopify Admin REST API Documentation
• Shopify Admin GraphQL API Reference
• Shopify Storefront API Guide
• Shopify OAuth and App Installation
• Shopify Webhooks Reference