Deploy class instance flow (Prefect 3)

Help
1

I am trying to deploy this simple class instance flow in a local setup (local worker, local server).

from prefect import flow

class MyClass:
    @flow
    def instance_method_flow(self):
        pass

What should I specify as an entrypoint?

The working directory is set correctly with pull/prefect.deployments.steps.set_working_directory/directory.

I have tried these, but none of them works:

  • MyClass.instance_method_flow
  • MyClass().instance_method_flow()
  • MyClass().instance_method_flow
  • instance_method_flow (prefect deploy finds out this)
  • I have even created a wrapper variable like this, and specified it as the entry function:
    mflow = MyClass().instance_method_flow()

Either way, Prefect could not find the flow on trying to run it.
Any suggestions?

2

I have managed to track this down. A record to myself and to others after that.

The problem is 2 fold.

  1. According to the documentation you can use class methods as flows, but you cannot deploy them, because the import_object function in prefect.utilities.importtools cannot handle it.
    A workaround is to put your class in a module and create a top level alias and reference that in the entrypoint. For example:
    pkg/class_flow.py: method_flow_alias = MyClass().method_flow
    Then in a deployment in prefect.yaml the entrypoint would be: pkg.class_flow:method_flow_alias
    Scripts (i.e. entrypoints with .py:) are parsed and loaded differently, and they cannot be used this way.
  2. The parameter schema of the deployment (check parameter_openapi_schema with prefect deployment inspect), will require your flow to get a self parameter. parameter_openapi_schema is generated with Pydantic based on the function signature (inspect.signature). As of Pydantic’s documentation the only way to make it optional if its default value is None.
    Therefore just add None as default value to your class method flow signature, and it will work.

Working minimal example:

# pkg/class_flow.py
from prefect import flow

class MyClass:
  @flow(log_prints=True)
  def instance_method_flow(self=None):
    print('INSTANCE METHOD FLOW')

method_flow_alias = MyClass().instance_method_flow

# prefect.yaml
deployments:
- name: method_depl
  entrypoint: pkg.class_flow:method_flow_alias
# ...

Test and usage:

prefect deploy -n method_depl
prefect deployment inspect "instance-method-flow/method_depl" | grep parameter_openapi_schema # "self" should not be in required
prefect deployment run "instance-method-flow/method_depl" # should be okay