if end_date_parameter:
        end_date = end_date_parameter
    else:
        end_date = pendulum.now(tz='Europe/London').to_date_string()

num_days_parameter = Parameter('num_days', default=1)
num_back_fill_days_parameter = Parameter('num_back_fill_days', default=1)
end_date_parameter = Parameter('end_date', default=None)

@task
def calculate_flow_end_date(end_date: str):
    if end_date is not None:
        return end_date
    return prefect.context.get('scheduled_start_time').to_date_string()

common_flow_result = common_flow(parameters={
    'num_days': num_days_parameter,
    'num_back_fill_days': num_back_fill_days_parameter,
    'end_date': task_end_date,
})

from prefect import Flow, Parameter, task
from prefect.tasks.shell import ShellTask


@task(log_stdout=True)
def log_param_value(user_input: str):
    print(user_input)


shell = ShellTask(return_all=True)

with Flow("shell-flow") as flow:
    param = Parameter("user_input", default="your_value")
    log_param_value(param)
    shell_task = shell(command=f"echo {param}")

if __name__ == '__main__':
    # flow.run()
    flow.run(parameters=dict(user_input="Hello World"))

ShellTask

log_param_value

StartFlowRun

ShellTask

@task(log_stdout=True)
def log_parameter_value(parameter_name: str, parameter_value: (str, int, float)):
    <http://logger.info|logger.info>('Parameter: (%s, %s)', parameter_name, parameter_value)


num_days_parameter = Parameter('num_days', required=True)
num_back_fill_days_parameter = Parameter('num_back_fill_days', required=True)
end_date_parameter = Parameter('end_date', required=True)

shell_command = ShellTask(
    name='served_advert_task',
    stream_output=True,
)

with Flow(
    'my_flow',
    storage=Docker(
        base_image='my-docker-image:latest',
        local_image=True,
    )) as flow:

    log_parameter_value('num_days', num_days_parameter)
    log_parameter_value('num_back_fill_days', num_back_fill_days_parameter)
    log_parameter_value('end_date', end_date_parameter)

    shell_command(
        command=construct_etl_command(),
    )

/tmp/prefect-qc0rm5qd: line 1: Parameter:: No such file or directory

shell_task2.set_dependencies(upstream_tasks=[shell_task1])

ShellTask

flow

f

with Flow(...) as flow:
     x = Paramater("x", default = 0)
     MyTask(...)(command=f(x())

x

from prefect import Flow, task, Parameter, Task
import prefect

class TestTask(Task):

    def run(self, x):
        logger = prefect.context.get("logger")
        <http://logger.info|logger.info>(x)
        return "test_" + x

test = TestTask()

def mycallable(x):
    return "call_" + x

with Flow("aaa") as flow:
    x = Parameter("x", default="x")
    test(mycallable(x))

flow.run()

with Flow("aaa") as flow:
    x = Parameter("x", default="x")
    TestTask()(mycallable(x))

flow.run()

import prefect
from prefect import (
    Flow,
    Parameter,
    task,
)
from prefect.storage import Docker
from prefect.tasks.shell import ShellTask


# @task(log_stdout=True)
def construct_etl_command(num_days: str = None) -> str:
    return 'echo {0}'.format(num_days)


@task(log_stdout=True)
def log_parameter_value(parameter_name: str, parameter_value: (str, int, float)):
    """
    Log the given parameter to the Prefect logger.
    :param: parameter_name: The name of the parameter.
    :param: parameter_value: The value of the parameter.
    """
    prefect_logger = prefect.context.get('logger')
    <http://prefect_logger.info|prefect_logger.info>('Parameter: (%s, %s)', parameter_name, parameter_value)


num_days_parameter = Parameter('num_days', required=True)

shell_task = ShellTask(name='shell_task', stream_output=True)

with Flow(
        'minimal_broken_docker_flow',
        storage=Docker(
            base_image='python:3.9',
        )) as flow:

    log_parameter_value('num_days', num_days_parameter)

    shell_command = construct_etl_command(
        num_days=num_days_parameter,
    )
    shell_task = shell_task(command=shell_command)

# @task(log_stdout=True)

@task