recipe_modules/cas_util/api.py

# Copyright 2021 The Fuchsia Authors. 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 CasUtilApi(recipe_api.RecipeApi):
    """API for archiving and accessing archived files in RBE-CAS."""

    def upload(
        self,
        staging_dir,
        upload_paths=(),
        step_name=None,
        output_property=None,
    ):
        """Returns the digest of the archived tree created from the provided
        staging_dir.

        Args:
            staging_dir (Path): The root of the directory to upload.
            upload_paths (list(Path)): A list of files under the staging_dir
                to upload if not uploading the entire directory.
            output_property (str): The name of the output property to set with
                the digest.
        """

        def do_upload(step_name):
            return self.m.cas.archive(
                step_name,
                staging_dir,
                *upload_paths,
                # Debug logs are helpful for understanding the root cause of
                # upload failures.
                log_level="debug",
                # TODO(fxbug.dev/122153): Stop setting a timeout once CAS
                # slowness is resolved and/or the CAS CLI sets shorter
                # internal timeouts.
                timeout=datetime.timedelta(minutes=15),
            )

        # TODO(fxbug.dev/62939): The two codepaths below mirror what was in
        # upload_isolated before the migration happened so that step names stay
        # the same. Combine the two paths once we confirm there are no
        # dependencies on the step names.
        if output_property:
            with self.m.step.nest(step_name or "cas") as step:
                cas_digest = do_upload("archive")
                step.presentation.properties[output_property] = cas_digest
        else:
            cas_digest = do_upload(step_name or "archive")
        return cas_digest

    def download(self, digest, output_dir, step_name=None):
        """Downloads the archive with the CAS digest into output_dir."""
        self.m.cas.download(
            step_name or "download", digest=digest, output_dir=output_dir
        )