I don't understand the proper use of the

context

dict. The documentation is not more than a hello-world. I'd like to use Prefect's Context to pass some (configuration) constants to task but that's not possible.

#!/usr/bin/env python
# coding: utf8

import prefect
from prefect import task, Flow
from prefect.environments.storage.local import Local


@task
def print_context():
    prefect.context.get("logger").info("get-val is '%s'", prefect.context.get("val"))
    prefect.context.get("logger").info("dot-val is '%s'", prefect.context.val)


with Flow("contexttest", storage=Local(directory="/flows/.prefect/flows")) as flow:
    with prefect.context(val="SPAM"):
        print_context()

if __name__ == "__main__":
    flow.register()

Will print

'None'

and in the second line throws an exception. Thus, the context's

val

is only valid in the

with

block. But what's the purpose of

Context

if not passing some simple constants around? I can also write

with Flow("contexttest", storage=Local(directory="/flows/.prefect/flows")) as flow:
    prefect.context.val="SPAM"
    print_context()

with the same result: not available in task.

Hello @Sven Teresniak - Context is mostly used for internal purposes, to set information and provide it to many consumers at once, but you can take advantage of it if it’s useful. However, I think you are confusing Prefect’s build-time behavior from its run-time behavior. When you enter your context, you’re doing so at build-time. Your task’s

run()

function won’t actually be called until later, when you (or an Agent) call

flow.run()

. At that time, the context will no longer be available. Therefore, if you want to combine a context at build-time with a task at run-time, you’ll need to write a task class that executes its

__init__()

within the context manager. Something like:

import prefect
from prefect import task, Flow, Task
from prefect.environments.storage.local import Local


@task
def print_context():
    prefect.context.get("logger").info("get-val is '%s'", prefect.context.get("val"))
    prefect.context.get("logger").info("dot-val is '%s'", prefect.context.val)

class PrintContext(Task):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.val = prefect.context.get('val')

    def run(self):
        prefect.context.get("logger").info("dot-val is '%s'", self.val)


with Flow("contexttest",) as flow:
    with prefect.context(val="SPAM"):
        PrintContext()()
flow.run()

This way, you can capture the information at build-time and re-use it at run-time. However, this is an unusual situation - if you have information that will vary at runtime, we suggest using a Parameter; if you have information available at build-time, we suggest passing it directly to the task as a contant.

@Jeremiah Thank you very much for the explanation! You're right. Storing sometimes-changing config stuff as instance variables in the Task is easy. I prefer the

@task

-decorated more but in this case its better to subclass

Task

. In general its maybe a good idea to not let people exchange data of any kind using the context. 🙂 Thanks again!

Haha to be completely honest I once led a crusade to not expose

context

to users at all but was rejected 😉 so instead I just show up every now and then to share examples like this one. Glad it’s working for you!