simple_download_with_annotation.py

import kfp
from kfp import dsl
from kfp import components
from kubernetes import client as k8s_client
import os

def produce_output(
        mydestfile: kfp.components.OutputPath(), # Note: _file is stripped
):
    import urllib.request

    print("starting download...")
    urllib.request.urlretrieve("http://212.183.159.230/20MB.zip", mydestfile)
    print("done")


produce_output_op = components.create_component_from_func(
    produce_output,
    base_image='quay.io/operate-first/opf-toolbox:v0.12.0',
)


@dsl.pipeline(
    name='simple_pipeline',
    description=''
)
def sdk_pipeline():
    import json
    produce_op = produce_output_op().add_pod_annotation(name='artifact_outputs', value=json.dumps(['mydestfile']))

from kfp_tekton.compiler import TektonCompiler
os.environ["DEFAULT_STORAGE_CLASS"] = os.environ.get(
    "DEFAULT_STORAGE_CLASS", "gp3"
)
os.environ["DEFAULT_ACCESSMODES"] = os.environ.get(
    "DEFAULT_ACCESSMODES", "ReadWriteOnce"
)
TektonCompiler().compile(sdk_pipeline,  __file__.replace('.py', '.yaml'))