Hi I m trying to implement a python library to gather all th Prefect Community #ask-community

Hi! I'm trying to implement a python library to ga...

Emma Rizzi

03/21/2022, 3:52 PM

Hi! I'm trying to implement a python library to gather all the tasks in common in my ETL flows and looking for some insights on the best way to do it with prefect. Sharing more in thread :

Emma Rizzi

03/21/2022, 3:53 PM

I have a generic DataManager (need for implementing subclasses) looking like:

Copy code

class DataManager():
    @task
    def download_data(self, source: str, destination: str):
        pass

And I tried to use that in a flow like:

Copy code

@task
def get_manager():
    return DataManager()

with Flow("Flow") as flow:
    manager = get_manager()
    manager.download_data(...)

This fails during build with

AttributeError: 'FunctionTask' object has no attribute 'download_data'

because of the object

manager

I understand this is because of my definition of the Manager object, and I could remove the

@task

from my library's functions, and use them as usual python function inside my flow But I'm not sure if there's already a standardized way to implement tasks in a python package ? I haven't found anything in the docs yet

Kevin Kho

03/21/2022, 3:57 PM

This is not because of the definition. This is because the

Copy code

with Flow...

builds the DAG but it’s not executed until

flow.run()

While it’s being build, the

get_manager

task hasn’t actually run so

manager

is of type

FunctionTask

(or Task) and doesn’t have the

download_data

attribute. You need to access

.download_data

inside a task to defer the execution of that to runtime also

Emma Rizzi

03/22/2022, 7:52 AM

Thanks @Kevin Kho does this mean download_data should not be a task as we could call it inside another task?

Anna Geller

03/22/2022, 10:45 AM

You can call a task within another task by calling its .run() method but this is generally not encouraged in most use cases. Can you define the problem that you try to solve? Do you try to reuse some functionality (your DataManager) across various flows?

Emma Rizzi

03/22/2022, 12:35 PM

@Anna Geller yes I try to gather some functionnalities with data managers (main objective is to have different implementations depending on the backend used, and abstract them for developpers) Is there a recommended way to implement Prefect Tasks in such cases, or should I stick to classic functions and call them inside Tasks ?

Anna Geller

03/22/2022, 12:49 PM

You are spot-on in that using functional API is the right way to go to build reusable tasks. This way, you could build those as building blocks that you can import and call within various flows. Also, using the functional API will make it much easier to migrate later to Prefect 2 (Orion) as the same syntax will to a large extent work the same way

Emma Rizzi

03/22/2022, 1:24 PM

Thanks! I'll keep working on that direction, we hope to try out orion soon as well 🙂

🙌 1

Kevin Kho

03/22/2022, 2:29 PM

More like you can do

Copy code

with Flow(...):
    SomeClass().sometask()

and this will work, but if you so

Copy code

with Flow(...):
    some = task_that_returns_SomeClass()
    some.sometask()

this will not because

some

is not evaluated until runtime so you have to do:

Copy code

@task
def do_sometask(x):
    return x.sometask()

with Flow(...):
    some = task_that_returns_SomeClass()
    result = do_sometask(some)

👌 1

24 Views

Open in Slack

Previous Next