https://prefect.io
Join Slack
<@ULVA73B9P> What's the Prefect 2 version of `defaults_from_attrs` for subclassing Tasks?
# ask-marvin
m
@Marvin What's the Prefect 2 version of 
defaults_from_attrs
for subclassing Tasks?
m
Oh, the joys of change. In Prefect 2.0, the 
defaults_from_attrs
method no longer exists. Instead, you'll have to use the 
__getattribute__
method in your Task subclass to achieve the same effect. It's a bit more work, but then again, what isn't? Here's an example:
Copy code
python
from prefect import Task

class MyTask(Task):
    def __init__(self, my_attr=None, **kwargs):
        self.my_attr = my_attr
        super().__init__(**kwargs)

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if attr:
            return attr
        else:
            return self.context.get(item)
In this example, if 
my_attr
is not set when the Task is initialized, it will look for 
my_attr
in the context when the Task is run. It's a bit like having your own personal assistant, always there to fetch things for you. Must be nice.
n
🙅
my understanding (IIRC) is that 
defaults_from_attrs
was a convenience thing since we always had to carry attrs from the _`__init__`_ into the 
run
method, but now tasks enter the engine via 
__call__
so its just not relevant the same way anymore
what are you trying to do?
m
@Nate So, to be specific, what I'm trying to do is take the 
ShellOperation
task and make a custom version for running some CLI tools that we have, and I'd like to do a few things with some arguments that get passed to it to create the command
We had a bunch of old Prefect 1 code that did that,.
that used 
ShellTask
and 
defaults_from_attrs
n
Copy code
from prefect_shell import ShellOperation
from pydantic import validator

class MyShellOp(ShellOperation):

    @validator('commands')
    def validate_commands(cls, commands: list[str]) -> list[str]:
        for command in commands:
            if command == "rm -rf /":
                raise ValueError("I'm calling the FBI")
            
        # modify commands here if needed
        
        return commands
    
if __name__ == "__main__":
    MyShellOp(commands=["rm -rf /"]).run()
something like that?
m
Right!
@Nate Would that work as-is? And still have all the normal 
task
args and defaults and stuff?
hrm.. 
OSError: source code not available
n
did you do it in ipython?
m
doing it in Jupyter, yes
n
that's bc you subclass a block in an interactive context, idk why that happens to be honest
😭 1
should work in a script, and yeah, you're just hanging a validator off the existing class, so everything should be the same, except any changes you make to the 
commands
in that validator
m
@Nate So any changes I want happen in the 
validator
?
n
yeah, if you're only wanting to do 
list[str] -> some_transform -> list[str]
on your commands then that should work fine if you want to change more about the ShellOperation methods work, you'd have to override them and make sure they play the same role
👍 1
m
@Nate What if I want to pass additional args? The main thing I want to do is pass some arguments and have the task construct the CLI command based on them.
n
Copy code
from prefect_shell import ShellOperation
from pydantic import Field, root_validator

def some_function(commands: list, foo: str) -> list:
    return commands + [f"echo 'foo is {foo}'"]

class MyShellOp(ShellOperation):
    
    extra_kwargs: dict = Field(default_factory=dict)

    @root_validator
    def validate_commands(cls, values: dict) -> dict:
        if extra_kwargs := values.get("extra_kwargs"):
            new_commands = some_function(
                commands=values.get("commands"),
                **extra_kwargs
            )
            values["commands"] = new_commands
        return values
    
if __name__ == "__main__":
    MyShellOp(
        commands=["echo 'Hello, World!'"],
        extra_kwargs={"foo": "bar"}
    ).run()
something like that?
Copy code
» python my_shell_op.py
18:30:52.811 | INFO    | prefect.MyShellOp - PID 42012 triggered with 2 commands running inside the '.' directory.
18:30:52.861 | INFO    | prefect.ShellProcess - PID 42012 stream output:
Hello, World!
foo is bar
18:30:52.862 | INFO    | prefect.ShellProcess - PID 42012 completed with return code 0.
if you dont like the validator stuff, you can always override the 
run
method and add this extra kwargs handling as methods of your subclass and then call them in your subclass' run method
m
@Nate Well, ideally something more like
Copy code
class MyShellOp(ShellOperation):
    @root_validator
    def validate_commands(cls, commands: list[str], x: str, y: int):
        if extra_kwargs := values.get("extra_kwargs"):
            new_commands = some_function(
                commands=values.get("commands"), **extra_kwargs
            )
            values["commands"] = new_commands
        return values


if __name__ == "__main__":
    MyShellOp(commands=["echo 'Hello, World!'"], x="w4rsrtwer", y=6).run()
n
the root validator is going to give you a dict of 
values
on the class, unfortunately that's not something you can change to affect the attrs on the class e.g. x, y would have to be actual fields on the class like 
extra_kwargs
from my example
m
I think the easiest thing then would just be to separate all the logic from the old task into a new task, and have it return a string, and then map 
ShellOperation
over the resulting strings
@Nate How do I use 
map
with a method on a Task? So like
Copy code
ShellOperation.map(
        commands=list_of_list_of_commands,
    ).run()
But I feel like that won't work lol
n
so if "go to definition" (whatever that is your editor) you'll see that ShellOperation is not a task, but a JobBlock, which is a specific type of block. blocks have no submit / map interface like tasks usually the answer to things like this is, "wrap it in a task"
👍 1
m
Ahhhh okay
Oh, perfect
so that's all I had to do in the first place
just put all the logic in the parent task that runs the Block
Sweet, thanks!
n
👍
6 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

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.

Powered by