Is there a canonical way or a canonical task that ...
# ask-community
s
Is there a canonical way or a canonical task that folks use for writing arbitrary tasks to flow logs? Like if I were to have a (built-in) task whose output I'd like to log somewhere, do people generally just write a trivial task that looks something like:
Copy code
from prefect.utilities.logging import get_logger
from prefect import task

@task
def log_value(val: str) -> None:
    logger = get_logger()
    <http://logger.info|logger.info>(val)
I'm going to need to do this in a lot of flows so I'd rather not rewrite a method/task like this over and over
Or is it an option to call the logger directly within the flow context? My assumption is that that's not a best practice
k
I just tested it, I think you are correct about the flow context logging. The logging doesn’t occur where you would expect it to. Seems like prefect does a first pass over the flow to do some initialization, and the logging occurs there, instead of during the actual execution. Example:
Copy code
import prefect
from prefect import Flow, Parameter, task
from prefect.executors import LocalDaskExecutor


@task()
def compute_bar(foo) -> str:
    return 'bar'


with Flow('foo') as flow:
    foo = Parameter('foo')
    prefect.context.get('logger').info('foo: %s', foo)

    bar = compute_bar(foo)
    prefect.context.get('logger').info('bar: %s', bar)


if __name__ == '__main__':
    flow.run(foo=1)
I get the following log output
Copy code
[2021-09-14 15:56:39-0600] INFO - prefect | foo: <Parameter: foo>
[2021-09-14 15:56:39-0600] INFO - prefect | bar: <Task: compute_bar>
[2021-09-14 15:56:39-0600] INFO - prefect.FlowRunner | Beginning Flow run for 'foo'
[2021-09-14 15:56:39-0600] INFO - prefect.TaskRunner | Task 'foo': Starting task run...
[2021-09-14 15:56:39-0600] INFO - prefect.TaskRunner | Task 'foo': Finished task run for task with final state: 'Success'
[2021-09-14 15:56:39-0600] INFO - prefect.TaskRunner | Task 'compute_bar': Starting task run...
[2021-09-14 15:56:39-0600] INFO - prefect.TaskRunner | Task 'compute_bar': Finished task run for task with final state: 'Success'
[2021-09-14 15:56:39-0600] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
which is to say that even if I log the result of a task within the flow block, I only ever see the “uncomputed” version of it. The logging occurs before the flow runs for real (and does not happen again)
👍 1
using a task should work
s
yes I was just wondering if there was an out-of-the-box that would do that! If not I'll probably just end up making my own custom task class so that I can call it from lots of different flows...though I suppose I could use the built-in
ShellTask
and just have it echo some input
k
Yes you are right @Sean Talia, you can also to
@task(log_stdout=True)
. which will let you log just using print statements. If you store your Flow as a script though, the logger is evaluated during runtime as opposed to build time, which is what @Kyle McChesney faced. This might allow you to get the logger in the global namespace if the logger is loaded during runtime and you instantiate it in the script.