recipe_modules/safe_cipd/api.py

# Copyright (c) 2026, the Dart project authors. Please see the AUTHORS file
# for details. All rights reserved. Use of this source code is governed by a
# BSD-style license that can be found in the LICENSE file.

import datetime
from recipe_engine import recipe_api


class SafeCipdApi(recipe_api.RecipeApi):

  def register(self,
               package_name: str,
               pkg_def: 'recipe_engine.cipd.PackageDefinition',
               output_path: 'recipe_engine.path.Path',
               tags: dict[str, str] = None,
               unique_tags: dict[str, str] = None,
               refs: list[str] = None,
               report_bcid: bool = True):
    """Safely builds and registers (uploads) a CIPD package.

    Args:
      package_name: The name of the package.
      pkg_def: A CIPD PackageDefinition object.
      output_path: The Path to the output zip file.
      tags: A dict of tags to apply.
      unique_tags: A dict of unique tags to verify before uploading.
      refs: A list of refs to apply.
      report_bcid: Whether to report to BCID.

    Returns:
      The newly registered pin if uploaded.
      None if skipped (already exists).
      Fails the recipe if any existing package has a different instance_id.
    """
    pin = self.m.cipd.build_from_pkg(pkg_def, output_path)
    instance_id = pin.instance_id

    already_existing_tags = []
    for key, value in (unique_tags or {}).items():
      tag_str = f'{key}:{value}'
      existing = self.m.cipd.search(package_name, tag_str, test_instances=[])
      if existing:
        if any(instance.instance_id != instance_id for instance in existing):
          raise self.m.step.StepFailure(
              f'upload failed: {package_name}:{tag_str} already exists with a different instance id'
          )
        already_existing_tags.append(tag_str)

    if already_existing_tags:
      self.m.step.empty(
          f'upload skipped: {package_name} already exists for unique tags: {already_existing_tags}'
      )
      return None

    all_tags = (tags or {}) | (unique_tags or {})

    @self.m.time.exponential_retry(2, datetime.timedelta(seconds=1))
    def _register():
      if self.m.runtime.is_experimental:
        self.m.step.empty('upload skipped: experimental build')
        return None
      pin = self.m.cipd.register(
          package_name, output_path, tags=all_tags, refs=refs)
      if report_bcid:
        sha256 = self.m.file.file_hash(output_path)
        self.m.bcid_reporter.report_cipd(sha256, pin.package, pin.instance_id)
      return pin

    return _register()