Is there a way to use a simple function in a flow, for example when I am going to call the function many times. I can easily make it work by making my function a 'task', but I can't see why it should have all the scheduling overhead:

@task
def get_host_from_kv(opts):
    hid = Schema({
        'host': str,
        Optional('identity', default=0): int
    }).validate(opts.kvopts)
    return HostParam(identity=hid['identity'], host=hid['host'])


with Flow("Query a single host") as flow:
    opts = Parameter('opts')

    result = winrm_powershell(host=get_host_from_kv(opts),
                              command=load_script(opts, 'win_ps_version.ps1'),
                              opts=opts)

Hm, you could handle this in a few ways: • you could package that method and treat it as you would other dependencies • build it into an extendable task class, which you can further extend downstream, something like this:

from prefect import Task

class KVHost(Task):
  def get_host_from_kv(self, opts):
    # your method logic here

  def run(self):
    super(KVHost).run()

class AnotherTask(KVHost):
  def run(self):
    host = self.get_host_from_kv()
    # do something with host
    super(AnotherTask).run()

Thanks for the hints @nicholas That's quite interesting.

For sure - to me this feels like less of a Prefect issue and more of a code organization/modularization issue; I think personal/org preference and what level of observability you need will largely influence how you want to handle something like this.

It's more of a prefect use as prefect is building the flow when it loads the file.

@Rob Fowler you're missing a run method on the

OptedTask

, which means

Useful

is calling the base

Task

class run method which doesn't do anything in this case because that's where task logic goes. Try something like this instead:

import prefect
from prefect import Flow, Task, Parameter


class OptedTask(Task):
    def __init__(self, opts=None, **kwargs):
        # This lets you instantitate opts at build time
        self.opts = opts
        super().__init__(kwargs)

    def get_host_from_kv(self):
        return self.opts["kwopts"]["host"]


class Useful(OptedTask):
    def run(self, opts):
        # This lets you instantiate opts at runtime
        self.opts = opts
        host = self.get_host_from_kv()
        return host


with Flow("Query a single host") as flow:
    opts_param = Parameter("opts", default={"kwopts": {"host": "host"}})

    apt = Useful()

    result = apt(opts=opts_param)

flow.run()

Awesome, thanks so much for this. I'll return the effort with some sort of license. You don't have a great use case of license for us. We are a cloud provider, I am using prefect to orchestrate the construction of custom compute environments for mid end and government customers that don't have the expertise to run their own infrastructure on prem or in the cloud.

@Rob Fowler I'd be happy to put you in touch with one of our account managers - they can work with you on a license that makes sense for your work and scale.

Yes, I am using the self.opts from the constructor,

from prefect import Task

class OptTask(Task):
    def __init__(self,  opts, **kwargs):
        self.opts = opts
        super().__init__(kwargs)

@Rob Fowler would you like me to put you in touch with one of our account managers as I mentioned above?

not at this stage.

Ok no worries! 🙂