Bulk Tagging Shopify Products with PHP: One Tag, Thousands of Products

The right approach is Shopify's Bulk Operations API — specifically bulkOperationRunMutation. You build a JSONL file with one line per product, stage it with Shopify, submit a single mutation, and let Shopify apply the tag across every product in the background. Your PHP application just needs to manage the job lifecycle.

This post walks through exactly that, using the same MySQL queue and cron pattern we covered previously. If you haven't read that post (see here), the short version is: pending jobs sit in a queue table, a cron job submits them, polls for completion, and processes the JSONL result.

The Tagging Mutation

Shopify's tagsAdd mutation is what we want here. It takes a resource GID and an array of tags to add, and it's smart about it — it won't add a duplicate if the tag already exists on the product. That makes it safe to re-run without worrying about creating a mess.

In a normal (non-bulk) context it looks like this:

mutation tagsAdd($id: ID!, $tags: [String!]!) {
    tagsAdd(id: $id, tags: $tags) {
        node {
            id
        }
        userErrors {
            field
            message
        }
    }
}

In bulk mode, this same mutation string gets passed to bulkOperationRunMutation, with the variables for each product call supplied per-line in a JSONL file. One line, one product. Shopify runs the mutation against each line in sequence.

Step 1: Build the JSONL File

Each line of the JSONL needs to contain the variables for one mutation call — in this case, the product GID and the tag to apply. You might be pulling your product list from your own database, a CSV, an ERP feed, whatever — it doesn't matter, as long as you end up with a list of Shopify product IDs.

/**
 * Builds a JSONL string where each line is the variable payload for one tagsAdd call.
 *
 * @param array  $product_ids  Array of Shopify product IDs (numeric or full GID format)
 * @param string $tag          The single tag to apply to all products
 * @return string              Raw JSONL content ready to upload
 */
function build_tag_jsonl($product_ids, $tag) {
    $lines = [];

    foreach ($product_ids as $product_id) {
        $lines[] = json_encode([
            'id'   => 'gid://shopify/Product/' . basename($product_id),
            'tags' => [trim($tag)]
        ]);
    }

    return implode("\n", $lines);
}

The basename() trick strips any path prefix, so it handles both raw numeric IDs and full GID strings coming in from different sources without you having to normalise them upstream. A habit worth keeping.

The resulting JSONL looks like this — one JSON object per line, no trailing comma, no array wrapper:

{"id":"gid://shopify/Product/7823001649234","tags":["sale-2026"]}
{"id":"gid://shopify/Product/7823001649235","tags":["sale-2026"]}
{"id":"gid://shopify/Product/7823001649236","tags":["sale-2026"]}
...

Step 2: Stage the Upload

Before you can reference the JSONL in a bulk mutation, Shopify needs you to upload it to their infrastructure via a staged upload. You request a signed URL, PUT the file to it, and get back a resourceUrl to pass into the mutation.

/**
 * Stages the JSONL content with Shopify and returns the resource URL.
 *
 * @param string $jsonl_content  The raw JSONL string
 * @param string $shop_domain
 * @param string $access_token
 * @return string                The resourceUrl to reference in the bulk mutation
 */
function stage_tag_jsonl($jsonl_content, $shop_domain, $access_token) {

    ###### 1. Request a staged upload target ######
    $stage_mutation = '
        mutation {
            stagedUploadsCreate(input: {
                resource: BULK_MUTATION_VARIABLES,
                filename: "product_tags.jsonl",
                mimeType: "text/jsonl",
                httpMethod: PUT
            }) {
                stagedTargets {
                    url
                    resourceUrl
                    parameters { name value }
                }
                userErrors { field message }
            }
        }
    ';

    $stage_response = shopify_graphql_request($shop_domain, $access_token, $stage_mutation);

    $user_errors = $stage_response['data']['stagedUploadsCreate']['userErrors'] ?? [];
    if (!empty($user_errors)) {
        throw new RuntimeException('Staged upload failed: ' . json_encode($user_errors));
    }

    $target = $stage_response['data']['stagedUploadsCreate']['stagedTargets'][0];

    ###### 2. PUT the JSONL to the signed URL ######
    $ch = curl_init($target['url']);
    curl_setopt_array($ch, [
        CURLOPT_CUSTOMREQUEST  => 'PUT',
        CURLOPT_POSTFIELDS     => $jsonl_content,
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_HTTPHEADER     => ['Content-Type: text/jsonl']
    ]);
    curl_exec($ch);
    $http_code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
    curl_close($ch);

    if ($http_code !== 200) {
        throw new RuntimeException("Staged file PUT failed with HTTP {$http_code}");
    }

    return $target['resourceUrl'];
}

Step 3: Submit the Bulk Mutation

With the file staged, you're ready to fire the bulk operation. The mutation string you pass in is the per-product tagsAdd call — Shopify maps each JSONL line's fields to the variables in that mutation automatically.

/**
 * Submits a bulkOperationRunMutation to apply a tag to a list of products.
 *
 * @param string $resource_url   The resourceUrl from the staged upload
 * @param string $shop_domain
 * @param string $access_token
 * @return string                The Shopify bulk operation GID
 */
function submit_tag_bulk_mutation($resource_url, $shop_domain, $access_token) {

    ###### The per-product mutation that Shopify runs for each JSONL line ######
    $per_product_mutation = '
        mutation tagsAdd($id: ID!, $tags: [String!]!) {
            tagsAdd(id: $id, tags: $tags) {
                node { id }
                userErrors { field message }
            }
        }
    ';

    ###### The bulk wrapper mutation ######
    $bulk_mutation = '
        mutation bulkOperationRunMutation(
            $mutation: String!,
            $stagedUploadPath: String!
        ) {
            bulkOperationRunMutation(
                mutation: $mutation,
                stagedUploadPath: $stagedUploadPath
            ) {
                bulkOperation {
                    id
                    status
                }
                userErrors { field message }
            }
        }
    ';

    $variables = [
        'mutation'         => $per_product_mutation,
        'stagedUploadPath' => $resource_url
    ];

    $response = shopify_graphql_request($shop_domain, $access_token, $bulk_mutation, $variables);

    $user_errors = $response['data']['bulkOperationRunMutation']['userErrors'] ?? [];
    if (!empty($user_errors)) {
        throw new RuntimeException('Bulk mutation submission failed: ' . json_encode($user_errors));
    }

    return $response['data']['bulkOperationRunMutation']['bulkOperation']['id'];
}

Wiring It Into the Queue

The entry point for the whole operation — whatever triggers the tagging job — builds the JSONL, stages it, and inserts a row into the queue. The cron job takes it from there.

/**
 * Entry point: queue a bulk tag job for a given list of product IDs.
 *
 * @param PDO    $pdo
 * @param string $shop_domain
 * @param array  $product_ids   Array of product IDs to tag
 * @param string $tag           The tag to apply
 * @return int                  The queue job ID
 */
function queue_product_tag_job($pdo, $shop_domain, $product_ids, $tag) {
    $creds        = get_shop_credentials($shop_domain);
    $jsonl        = build_tag_jsonl($product_ids, $tag);
    $resource_url = stage_tag_jsonl($jsonl, $shop_domain, $creds['access_token']);

    $stmt = $pdo->prepare(
        "INSERT INTO shopify_bulk_jobs
            (shop_domain, operation_type, job_type, input_file_url)
         VALUES
            (:shop, 'mutation', 'product_tag', :input_url)"
    );
    $stmt->execute([
        'shop'      => $shop_domain,
        'input_url' => $resource_url
    ]);

    return $pdo->lastInsertId();
}

The staging happens at queue time, not at submission time. This keeps the cron script simple — by the time it picks up the job, the file is already on Shopify's infrastructure waiting. The cron just needs to call submit_tag_bulk_mutation() with the stored input_file_url.

The Cron Job

The cron picks up pending jobs, submits them, polls submitted ones, and processes completed ones. The submit step for a product_tag job just calls submit_tag_bulk_mutation() with the staged URL already in the queue row.

#!/usr/bin/env php
/**
 * cron_bulk_jobs.php
 * */5 * * * * /usr/bin/php /var/www/app/cron/cron_bulk_jobs.php >> /var/log/shopify_bulk.log 2>&1
 */

require_once __DIR__ . '/../bootstrap.php';

$pdo = get_db_connection();

###### Guard: bail if another instance is still running ######
$lock_file = '/tmp/shopify_bulk_cron.lock';
if (file_exists($lock_file) && (time() - filemtime($lock_file)) < 300) {
    echo "[SKIP] Previous run still active.\n";
    exit(0);
}
touch($lock_file);

###### 1. Submit pending jobs (one per shop at a time) ######

$pending = $pdo->query(
    "SELECT sbj.*
       FROM shopify_bulk_jobs sbj
      WHERE sbj.status = 'pending'
        AND NOT EXISTS (
            SELECT 1 FROM shopify_bulk_jobs active
             WHERE active.shop_domain = sbj.shop_domain
               AND active.status = 'submitted'
        )
      ORDER BY sbj.created_at ASC"
)->fetchAll(PDO::FETCH_ASSOC);

foreach ($pending as $job) {
    $creds = get_shop_credentials($job['shop_domain']);

    try {
        $bulk_id = submit_tag_bulk_mutation(
            $job['input_file_url'],
            $job['shop_domain'],
            $creds['access_token']
        );

        $pdo->prepare(
            "UPDATE shopify_bulk_jobs
                SET status = 'submitted', shopify_bulk_id = :bulk_id
              WHERE id = :id"
        )->execute(['bulk_id' => $bulk_id, 'id' => $job['id']]);

        echo "[INFO] Submitted job #{$job['id']} (product_tag) for {$job['shop_domain']}\n";

    } catch (Exception $e) {
        mark_job_failed($pdo, $job['id'], $e->getMessage());
    }
}

###### 2. Poll submitted jobs ######

$submitted = $pdo->query(
    "SELECT * FROM shopify_bulk_jobs WHERE status = 'submitted'"
)->fetchAll(PDO::FETCH_ASSOC);

foreach ($submitted as $job) {
    $creds  = get_shop_credentials($job['shop_domain']);
    $status = poll_bulk_operation(
        $job['shopify_bulk_id'],
        $job['shop_domain'],
        $creds['access_token']
    );

    if ($status['status'] === 'COMPLETED') {
        $pdo->prepare(
            "UPDATE shopify_bulk_jobs
                SET status = 'completed', result_url = :url
              WHERE id = :id"
        )->execute(['url' => $status['url'], 'id' => $job['id']]);

        echo "[INFO] Job #{$job['id']} completed. Result URL stored.\n";

    } elseif ($status['status'] === 'FAILED') {
        mark_job_failed($pdo, $job['id'], 'Shopify reported FAILED status');
    }
    // RUNNING / CANCELING: do nothing, wait for the next cron tick
}

###### 3. Process completed jobs ######

$completed = $pdo->query(
    "SELECT * FROM shopify_bulk_jobs WHERE status = 'completed'"
)->fetchAll(PDO::FETCH_ASSOC);

foreach ($completed as $job) {
    $pdo->prepare("UPDATE shopify_bulk_jobs SET status='processing' WHERE id=:id")
        ->execute(['id' => $job['id']]);

    try {
        process_tag_result($job);
        $pdo->prepare("UPDATE shopify_bulk_jobs SET status='done' WHERE id=:id")
            ->execute(['id' => $job['id']]);
        echo "[INFO] Job #{$job['id']} fully processed.\n";

    } catch (Exception $e) {
        mark_job_failed($pdo, $job['id'], $e->getMessage());
    }
}

unlink($lock_file);
echo "[DONE] Cron run complete.\n";

Processing the Result

Once the bulk operation completes, Shopify gives you a JSONL result file where each line is the response from one tagsAdd call. For a tagging job the success records are mostly just confirmation — what you actually want to catch are userErrors, which flag any products that couldn't be tagged (permissions issues, invalid GIDs, that sort of thing).

/**
 * Streams the bulk result JSONL and logs any per-product errors.
 *
 * @param array $job  The queue row from shopify_bulk_jobs
 */
function process_tag_result($job) {

    if (empty($job['result_url'])) {
        // No result URL means zero records were processed (empty JSONL, or all skipped).
        // Not necessarily an error — log it and move on.
        error_log("[WARN] Job #{$job['id']}: no result URL. Possibly empty input.");
        return;
    }

    $handle = fopen($job['result_url'], 'r');
    if (!$handle) {
        throw new RuntimeException("Could not open result URL for job #{$job['id']}");
    }

    $success_count = 0;
    $error_count   = 0;

    while (($line = fgets($handle)) !== false) {
        $record = json_decode(trim($line), true);
        if (empty($record)) continue;

        if (!empty($record['tagsAdd']['userErrors'])) {
            foreach ($record['tagsAdd']['userErrors'] as $err) {
                error_log("[TAG ERROR] Field: {$err['field']} — {$err['message']}");
            }
            $error_count++;
        } else {
            $success_count++;
        }
    }

    fclose($handle);
    error_log("[INFO] Job #{$job['id']}: {$success_count} tagged OK, {$error_count} errors.");
}

The Poll Function

Same as in any bulk operation setup — query currentBulkOperation and check the status. One thing to be aware of: Shopify's currentBulkOperation returns the most recently submitted operation for the shop. If you're running multiple shops through the same cron, verify the returned ID matches the one you're expecting before acting on it.

/**
 * Polls Shopify for the current bulk operation status.
 *
 * @return array ['status' => string, 'url' => string|null]
 */
function poll_bulk_operation($expected_bulk_id, $shop_domain, $access_token) {
    $query = '
        query {
            currentBulkOperation {
                id
                status
                errorCode
                objectCount
                url
            }
        }
    ';

    $response = shopify_graphql_request($shop_domain, $access_token, $query);
    $op       = $response['data']['currentBulkOperation'] ?? null;

    if (empty($op) || $op['id'] !== $expected_bulk_id) {
        // Either no operation running, or a different one — treat as unknown
        return ['status' => 'UNKNOWN', 'url' => null];
    }

    return [
        'status'       => $op['status'],
        'url'          => $op['url'] ?? null,
        'object_count' => $op['objectCount'] ?? 0
    ];
}

function mark_job_failed($pdo, $job_id, $message) {
    $pdo->prepare(
        "UPDATE shopify_bulk_jobs
            SET status = 'failed', error_message = :msg
          WHERE id = :id"
    )->execute(['msg' => $message, 'id' => $job_id]);
    error_log("[ERROR] Job #{$job_id} failed: {$message}");
}

A Few Things Worth Noting

Staged uploads expire. Shopify's staged upload URLs are time-limited. If you build the JSONL and stage it, then leave the job sitting in the queue for a long time before the cron submits it, the staged file may no longer be valid. In practice this isn't an issue if your cron runs every few minutes and you're not queuing jobs days in advance — but it's worth knowing.

tagsAdd is additive, not destructive. Applying a tag via this mutation will never remove existing tags. If you're running a seasonal tagging operation and want to clean up old tags first, that's a separate tagsRemove bulk job. Don't try to combine them into the same JSONL.

One bulk operation per store at a time. The NOT EXISTS clause in the pending query handles this, but it's worth making explicit: if you queue two tagging jobs for the same shop in quick succession, the second one sits in pending until the first reaches done. This is the correct behaviour — Shopify will reject the second submission anyway if the first is still running.

Empty result URL is valid. If Shopify had nothing to process — because your JSONL was somehow empty, or all lines were skipped — currentBulkOperation will return a COMPLETED status but with a null url. Handle that case in your result processing, as the code above does.

Wrapping Up

Applying a tag to a few hundred products feels like it should be trivial. It almost is — but doing it at scale, reliably, with full visibility over what succeeded and what didn't, needs a bit of structure around it. The bulk operations API is the right tool; the MySQL queue and cron pattern gives you the orchestration layer to use it safely.

The same pattern extends cleanly to tagsRemove if you need to strip tags, or to any other mutation that takes a single resource GID and a simple input — productUpdate for setting metafields in bulk, for example. Build the JSONL, stage it, submit, poll, process. Every time.