Sean Talia
09/14/2021, 9:34 PMfrom 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)
Sean Talia
09/14/2021, 9:35 PMSean Talia
09/14/2021, 9:36 PMKyle McChesney
09/14/2021, 9:56 PMimport 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)
Kyle McChesney
09/14/2021, 9:57 PM[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
Kyle McChesney
09/14/2021, 9:58 PMKyle McChesney
09/14/2021, 9:58 PMSean Talia
09/14/2021, 10:19 PMShellTask
and just have it echo some inputKevin Kho
@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.