Rob Fowler
11/05/2020, 10:25 PM@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)
nicholas
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()
Rob Fowler
11/05/2020, 11:34 PMnicholas
Rob Fowler
11/06/2020, 3:49 AMclass OptedTask(prefect.Task):
def __init__(self, opts, **kwargs):
self.opts = opts
super().__init__(kwargs)
def get_host_from_kv(self, opts):
return opts.kwopts.host
class Useful(OptedTask):
def run(self):
host = self.get_host_from_kv(self.opts)
return super().run(host)
with Flow("Query a single host") as flow:
opts = Parameter('opts')
apt = Useful(opts())
result = apt()
nicholas
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()
Rob Fowler
11/06/2020, 10:46 PMnicholas
init
with self.opts
you can ostensibly use the class without passing opts
again, but the most valuable use is the availability of self.opts
in other class methods without having to pass around the variable as an argument.Rob Fowler
11/09/2020, 10:02 PMfrom prefect import Task
class OptTask(Task):
def __init__(self, opts, **kwargs):
self.opts = opts
super().__init__(kwargs)
nicholas
Rob Fowler
11/10/2020, 10:09 PMnicholas