Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hello. I have a little question:
I found an example: <https://github.com/PrefectHQ/prefect/blob/master/examples/old/task_library/mysql/mysql_flow.py>
how I can make a host as a parameter?

Hi <@U01PK3GSH0S>, you want to make the `HOST` variable a Prefect `Parameter`. Is that right?

The situation here is that the task gets initialized before the `Flow` runs. Only the `task.run` method will be able to take `Parameters` . If you look at the code behind the task, some of the attributes can be overwritten during runtime. Unfortunately for this Task, the `HOST` can not be overwrriten. This is a simple fix of editing the Task definition to allow `HOST` to be overwrriten and if would be a welcome Pull Request (if you’re interested). Otherwise, the current workaround is something like the following:
```from prefect import Flow, Task, Parameter, task


class TestTask(Task):

    def __init__(self, p: str, **kwargs):
        self.p = p
        super().__init__(**kwargs)

    def run(self):
        self.logger.warning(f"Printing the parameter {self.p}")

@task
def override_task(override_p):
    task = TestTask(override_p)
    task.run()
    return

with Flow('test') as flow:
    p = Parameter('p', default="aaa")
    override_task(p)

flow.run(parameters=dict(p="testing"))```

You need to make a side `Task` that instantiates and calls the run method of the `MySQL` task. This override_task can now take the `Parameter`.

Hey one more time <@U01QEJ9PP53>,

I would like to ask about MySQL task implementation.

I have tested your  provided code and it works good - parameters are achievable but it seems that MySQL task is not executed(just only creating task instance but not getting MySQL response or error code if it is not successfully executed).

I believe, that the problem is behind my code:

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



class TestTask(Task):

    def __init__(self, p: str, **kwargs):
        self.p = p
        super().__init__(**kwargs)

    def run(self):
        mysql_response = MySQLExecute(
            name="testing_connection",
            db_name="",
            user="",
            host=self.p,
            password='',
            port=1234
        )
        self.logger.warning(f"Printing the parameter {self.p}")
        return mysql_response

@task
def override_task(override_p):
    task = TestTask(override_p)
    return task.run()

with Flow('test') as flow:
    p = Parameter('p', default="aaa")
    override_task(p)

flow.run(parameters=dict(p="host_name"))```
Even declaring empty, incorrect values for db password, user, db_name, etc, I got my tasks run with no error:
```[2021-04-08 17:21:32+0300] INFO - prefect.TaskRunner | Task 'override_task': Finished task run for task with final state: 'Success'```
Could you elaborate more on how to implement MySQL execution, because I'm clearly  misunderstanding this case.

Yeah you want to call the `MySQLExecute.run()` because this code is just initializing it. The run method is the one that will execute it

In the `override_task`, just put the MySQLExecute task in there instead of `TestTask` and then call the `run()` and that should work