Feature: Support spinning a single metaflow step

[talsperre]

To test spin on a new flow you can do the following:

Simple case:

python complex_dag_flow.py --environment=conda spin step_d

Pass in specific pathspec:

python runtime_dag_flow.py --environment=conda spin --spin-pathspec RuntimeDAGFlow/13/step_c/275232971

Pass in custom artifacts via module:

python runtime_dag_flow.py spin --spin-pathspec RuntimeDAGFlow/13/step_d/275233082 --artifacts-module ./my_artifacts.py

Skip decorators (including the whitelisted ones):

python complex_dag_flow.py --environment=conda spin step_d --skip-decorators

Use with Runner API:

    with Runner('complex_dag_flow.py', environment="conda").spin(
        step_name,
        spin_pathspec="<Some Val>",
        artifacts_module='./artifacts/complex_dag_step_d.py',
    ) as spin:
        print("-" * 50)
        print(f"Running test for step: step_a")
        spin_task = spin.task
        print(f"my_output: {spin_task['my_output']}")
        assert spin_task['my_output'].data == [-1]

See the tests for more examples on hot to use this command.

[Copilot]

Pull Request Overview

This PR introduces support for spinning a single Metaflow step. Key changes include new tests for spin functionality, the addition of spin‐related options in CLI and runtime components, and updates to datastore handling to support spin metadata and artifacts.

Reviewed Changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated no comments.

File	Description
test/unit/spin/*	Adds tests to exercise spin functionality across various scenarios
metaflow/*	Introduces new functions and parameters (e.g. spin_metadata, spin_artifacts) and a new SpinRuntime class to support the spin command
metaflow/cli_components/*	Updates CLI command definitions to include spin and spin-step options, along with handling for artifacts modules and decorator skipping
metaflow/datastore/*	Adds support for spin datastores and adjusts task datastore retrieval with spin-related parameters

Comments suppressed due to low confidence (2)

metaflow/runtime.py:712

• Consider replacing the print statement with a logger call (for example, a debug-level logger) so that timing metrics are recorded according to the system's logging configuration.

print(f"Time taken to run the step function: {time.time() - st_time:.2f} seconds")

metaflow/runtime.py:773

• Replace this print statement with a logging call to integrate timing information into the structured logging system and avoid reliance on direct console output.

print(f"Time taken to persist the output: {time.time() - st_time:.2f} seconds")