https://prefect.io
Join Slack
is there a way to initialise a custom result handler in a prefect agent run flow? I can, of course, ...
# ask-community
r
is there a way to initialise a custom result handler in a prefect agent run flow? I can, of course, initialise a result handler at build time but now I am all fancy with common containers for test and production I don't want the redis/azure result destination to be hard coded in the docker storage.
It does pickle the connection but I'd like it to call the 
__init__
at run time in the agent. I could further hack it, like it does, by calling it if it is un-initialised but not going through any 'init' seems it might lead to other troubles.
k
Hey @Rob Fowler, just want to make sure I understand the problem you’re trying to solve. You need different ResultHandlers for test and prod environments because of different locations for Redis and Azure results?
What Prefect version are you using?
r
master head
yes, I am using the same image but would like different redis servers. I'd like it to be dynamic, ideally based ont he environment so I can inject it
k
I’ll float 2 ideas here. One is environment variables on the agents for different environments. The second idea is as a parameter that you pass in during the run. What are your thoughts?
Getting confirmation from the team that the parameter would work. Agent environment variables will be more certain.
The parameter to populate the 
Result
location should work.
r
ps, in master as the register and build are not released yet. No an environment variable does not work. The 
__init__
method is not called at all. But, I am not sure how it can be as the init is not in the flow.
Copy code
import os
import sys

from prefect import Flow, Parameter

sys.path.append(os.path.abspath(os.getcwd()))  # noqa E402

from ext.redis_result import RedisResult
from tasks.ptest import test_task # noqa

with Flow("ptest-flow") as flow:
    opts = Parameter('opts', default={})
    result = test_task(opts)


from mcs_env.custom_docker_storage import ComplexDocker

# 'loading_flow': True,
# 'local_script_path': 'ptt/ptest.py',

print("========================= HERE?")
flow.storage = ComplexDocker(image_name="ptest")
flow.result = RedisResult()
flow.storage.build()
Maybe just a hint as to where I can init my result handler and have it called when the flow is run in prefect.
It won't be called in the with Flow(..) line as that has already run. Obviously not later in this file, as that has run too. I can have it init locally at build time either way here. As the flow is pickled. But the server variable in the environment is pickled in. (Which does work but I want it set in the agent like PrefectSecret or the Redis tasks etc)
As it is infra I am averse to providing a parameter on every run. I can also re-call 
__init__
automatically during the run if the redis server is not set at runtime and this will read the environment I have injected into the agent. This works but it's a bit of a hack.
z
Hey @Rob Fowler -- have you read https://docs.prefect.io/core/concepts/templating.html and seen if that fits your use-case?
You could also use script based storage instead of pickle based if you want things to be executed at runtime but we do try to expose dynamic values where necessary since that's a tricky thing to depend on.
r
moving to script storage would also work. Maybe the pushback here is that this is something that cloud already does?
It seems to be that results, like storage, being a flow parameter is a special case
z
I'm a bit confused. Are you trying to template something other than the result location?
Templating the result location isn't a Cloud-only feature.
r
No, I want to use a persistent result store in server
that's all
at flow level
only local results are supported for server and that needs no initialisation
I simply don't have a place to call the common initialiser with pickled flows.
z
You should be able to use any result type in Server, perhaps I'm misunderstanding you? You've run into one of the limitations of pickle based storage as far as having a initializer run at flow start. We are considering moving away from recommending it because pickling can be finicky. One solution is to have a callback on flow startup, you can open a feature request for that in the core repo if you'd like -- I'm not sure what the implications would be off the top of my head.
r
a callback with a list of initialisers would be the go
I think I said that, or thought it 🙂
or a decorator that adds and initialiser like in flask?
z
If you want that, it'd be nice to see a fleshed out request where you've thought what you're looking for through. It's pretty hard to know what you want from this thread 🙂 That said, I really do think templating with a context variable should work just fine with 
flow.result = RedisResult(location="{foo}/{task_run_id}")
then just provide 
PREFECT__CONTEXT__FOO
in your flow run environment.
r
with script encoded flows I assume.
z
With pickle too
Notice the lack of the 
f
string formatter. We're delegating the formatting to the flow runner which templates result locations as in the doc I linked.
r
if that works I am fine, I can't see how it can be run though, from my example above.
as it stands my RedisResult() uses an environment variable to configure it the same as azure. But the picked flow only pickles the flow in the context.
Maybe I am still not communicating what is wrong effectively. Sorry.
Simply put, for prefect server, if the flow uses a results object at the flow level (in 
with Flow("ptest-flow", result=CustomResult()) as flow:
CustomResult.__init__(..
is never called at runtime.
z
If you subclass the 
Result
type you should use the "approved" technique of templating the location (which can use environment variables via the Prefect context)
Or just use script based storage and this problem will go away -- there's not usually a good argument for using pickle based storage.
r
that is what I do, my subclassed RedisResult is just the same as azure and gcs
I'll switch to script based and see how I go.
I agree, I think the main issue here is, pickling does not call the init( at runtime.
z
Yeah I see the AzureResult type uses an environment variable -- that's poorly done because it won't work in pickle based storage if you're trying to set the value dynamically (as you've encountered). The solutions are the same as I've outlined.
r
script based deployment works.
👍 1
although I made a switch for build and run time:
Copy code
if 'local_script_path' in context:
    from mcs_env.custom_docker_storage import ComplexDocker

    flow.storage = ComplexDocker(image_name="ptest", flow_file="ptest-flow.py")
    flow.storage.build()
~
Seems now when it scans the flows directory it does a build for all flows as that is what the code says. I need some sort of conditional in the context to tell the module when and when not to build a docker container.
aha, OK. Seems in the build mode it calls the .build() itself. That is why I was seeing all the extra builds.
3 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