https://prefect.io
Join Slack
Hi there - I have a flow where I would like to (at flow run time) get a date (as a string) from a fu...
# ask-community
k
Hi there - I have a flow where I would like to (at flow run time) get a date (as a string) from a function defined in my codebase and provide it to ALL of the tasks in the flow. This sounds like a Parameter, but there are a couple problems with this: 1. a function call is not JSON serializable 2. I don’t really want every single task to depend on one node that is just getting a string - the graph view would be extremely cluttered Any ideas on how to accomplish this?
k
Hey @Kevin Weiler, you can: 1. Create a task to return the date. The parameter can be something like “today” or “days_from_today=0", then the task will calculate it and return it. 2. Maybe pull it somehow from the prefect context 3. Store your flow as a script and calculate it at the top (like after imports). Use the variable in your tasks directly (I think this will work). Docs on pickle vs flow based storage
k
thanks @Kevin Kho - if I do #1 - every task that needs this date will have a graph connection the the parameter and the downstream task right?
k
Context has `prefect.context.get(“today”). See this .
Yep that’s right
k
ok - I think i’ll give the script storage a try - I’m using Docker Storage so it looks like that is supported
that’s gonna need a big refactor I guess though because I’m currently using the imperative API and some builder functions to build the flow. I guess I could generate the script with a Jinja Template
k
Maybe pulling it from context is the easiest then?
k
yeah - having a look The story here is that we have a flow we want to run once per day, but it might take longer than a day to complete. I could tell each individual task to ask what day it is - but if it goes past 24 hours - it’s going to run flow steps with the wrong date
k
The context won’t change once the flow is running, but there is also 
scheduled_start_time
k
Does this context time return in the system’s time zone or UTC?
k
Ah good question. It might be UTC. Let me check
It is UTC
k
ok - I actually think it’s a moot point because we’ll be running at 4PM Chicago time - which is variously 21 or 22 UTC, but the same date
if you re-run a flow from failed, do you get a new context?
k
all of the fields are the same except the ones around timestamps which change. 
scheduled_start_time
would be the same though
this is the list of things that change
k
cool yeah, so 
scheduled_start_time
it is
somewhat separately - is this documentation correct? https://docs.prefect.io/api/latest/utilities/context.html I can’t seem to import 
prefect.context
…
I’m on 
0.14.7
but the 
0.14.22
documentation looks to be the same
k
You 
import prefect
and then
Copy code
prefect.context.get("scheduled_start_time")
inside the task
k
do you do the 
import
in the task itself also?
k
no need. import at the top of the script.
🙏 1
Copy code
import prefect
from prefect import task, Flow


@task()
def log_stuff(x):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(prefect.context.get("scheduled_start_time"))
    return x
k
yeah - i think that doc is incorrect - you can’t
Copy code
import prefect.context
just an FYI - nbd
k
Ah crud I see your point.
@Marvin open “Context documentation incorrectly has ‘import prefect.context’ which is invalid”
m
k
Thanks for flagging that!
k
Thanks so much, this works as expected
9 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