basic containertask example

container_task.py

"""A simple Flyte example."""

import typing
from flytekit import ContainerTask, dynamic, kwtypes
import yaml
import json

register = ContainerTask(
    name="register",
    input_data_dir="/var/inputs",
    output_data_dir="/var/outputs",
    inputs=kwtypes(git_url=str, git_commit_target=str,flyte_config_yaml=str,  im_registry=str, im_user=str, im_pass=str),
    outputs=kwtypes(output=str),
    image="ghcr.io/zeryx/flytekit:remote-register",
    command=[
        "sleep",
        "100000000"
        # "./execute.sh",
        # "{{.inputs.git_url}}",
        # "{{.inputs.git_commit_target}}",
        # "{{.inputs.flyte_config_yaml}}",
        # "{{.inputs.im_registry}}",
        # "{{.inputs.im_user}}",
        # "{{.inputs.im_pass}}",
        # "/var/outputs"
    ]
)


@dynamic
def proxy_registration_wf(git_url: str, git_commit_target: str = "master") -> str:
    """This is a dynamic task that will be executed at runtime."""
    with(open("config.json", 'r')) as f:
        config = json.load(f)
    im_registry = config['im_registry']
    im_user = config['im_user']
    im_pass = config['im_pass']

    flyte_config = config['flyte']
    flyte_config_yaml = yaml.dump(flyte_config)
    return register(
        git_url=git_url,
        git_commit_target=git_commit_target,
        flyte_config_yaml=flyte_config_yaml,
        im_registry=im_registry,
        im_user=im_user,
        im_pass=im_pass
    )