liren zhang
12/10/2020, 6:01 PMimport prefect
import prefect.utilities.context
from prefect.environments.storage import Local
from prefect.engine.results import LocalResult
from prefect import task, Flow
with prefect.context(a=1, b=2):
print(prefect.context.a) # 1
@task(name='task_today')
def report_start_day():
logger = prefect.context.get("logger")
<http://logger.info|logger.info>(prefect.context.today)
<http://logger.info|logger.info>(prefect.context.flow_run_id)
with Flow('My flow', storage=Local(), result=LocalResult()) as flow:
report_start_day()
flow.register(project_name="prefect_first_project")
Zanie
LocalResult
is for the return value from the task. The Prefect logger doesn’t write to a file by default, you can attach another stream handler to the logger that writes to a local file if you’d like the thoughIn [1]: from logging import getLogger, FileHandler
In [2]: logger = getLogger("prefect")
In [3]: logger.addHandler(FileHandler("foo.log"))
In [4]: import prefect
...: import prefect.utilities.context
...: from prefect.environments.storage import Local
...: from prefect.engine.results import LocalResult
...: from prefect import task, Flow
...: with prefect.context(a=1, b=2):
...: print(prefect.context.a) # 1
...: @task(name='task_today')
...: def report_start_day():
...: logger = prefect.context.get("logger")
...: <http://logger.info|logger.info>(prefect.context.today)
...: <http://logger.info|logger.info>(prefect.context.flow_run_id)
...: with Flow('My flow', storage=Local(), result=LocalResult()) as flow:
...: ^Ireport_start_day()
...: flow.run()
1
[2020-12-10 12:14:45] INFO - prefect.FlowRunner | Beginning Flow run for 'My flow'
[2020-12-10 12:14:45] INFO - prefect.TaskRunner | Task 'task_today': Starting task run...
[2020-12-10 12:14:45] INFO - prefect.task_today | 2020-12-10
[2020-12-10 12:14:45] INFO - prefect.task_today | ac1aff53-37c5-4a80-8044-b9009827cdbc
[2020-12-10 12:14:45] INFO - prefect.TaskRunner | Task 'task_today': Finished task run for task with final state: 'Success'
[2020-12-10 12:14:45] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
Out[5]: <Success: "All reference tasks succeeded.">
In [5]:
Do you really want to exit ([y]/n)?
❯ cat foo.log
Beginning Flow run for 'My flow'
Task 'task_today': Starting task run...
2020-12-10
ac1aff53-37c5-4a80-8044-b9009827cdbc
Task 'task_today': Finished task run for task with final state: 'Success'
Flow run SUCCESS: all reference tasks succeeded
liren zhang
12/10/2020, 7:22 PMZanie
logger.addHandler(FileHandler("foo.log"))
from within your taskliren zhang
12/10/2020, 8:28 PMprefect agent local start --show-flow-logs
and thought this is exactly what I need; but I will need to be able to do this in pythonZanie
❯ cat ~/prefect/scripts/example-log-locally.py
import os
from logging import FileHandler
import prefect
from prefect.environments.storage import Local
from prefect import task, Flow
LOG_PATH = os.path.expanduser("~/flow-that-logs.log")
def get_logger():
logger = prefect.context.get("logger")
logger.addHandler(FileHandler(LOG_PATH))
return logger
@task()
def report_start_day():
logger = get_logger()
<http://logger.info|logger.info>(prefect.context.today)
<http://logger.info|logger.info>(prefect.context.flow_run_id)
with Flow('flow-that-logs', storage=Local()) as flow:
report_start_day()
flow.register(project_name="first")
❯ python example-log-locally.py
Result check: OK
Flow URL: <https://cloud.prefect.io/michael-prefect-personal/flow/fdef6d85-4ebe-467f-bdcf-911ed50cfebc>
└── ID: 9d8e4451-b6db-4726-b9d8-720b2c2d877c
└── Project: first
└── Labels: ['mzmbp.local']
❯ prefect agent local start
____ __ _ _ _
| _ \ _ __ ___ / _| ___ ___| |_ / \ __ _ ___ _ __ | |_
| |_) | '__/ _ \ |_ / _ \/ __| __| / _ \ / _` |/ _ \ '_ \| __|
| __/| | | __/ _| __/ (__| |_ / ___ \ (_| | __/ | | | |_
|_| |_| \___|_| \___|\___|\__| /_/ \_\__, |\___|_| |_|\__|
|___/
[2020-12-10 21:22:47,646] INFO - agent | Starting LocalAgent with labels ['mzmbp.local', 'azure-flow-storage', 'gcs-flow-storage', 's3-flow-storage', 'github-flow-storage', 'webhook-flow-storage', 'gitlab-flow-storage']
[2020-12-10 21:22:47,646] INFO - agent | Agent documentation can be found at <https://docs.prefect.io/orchestration/>
[2020-12-10 21:22:47,646] INFO - agent | Agent connecting to the Prefect API at <https://api.prefect.io>
[2020-12-10 21:22:47,939] INFO - agent | Waiting for flow runs...
[2020-12-10 21:23:06,457] INFO - agent | Found 1 flow run(s) to submit for execution.
[2020-12-10 21:23:06,905] INFO - agent | Deploying flow run 10377b6e-3d59-4447-8692-db101134ee12
❯ cat ~/flow-that-logs.log
2020-12-10
10377b6e-3d59-4447-8692-db101134ee12
liren zhang
12/11/2020, 12:51 AMBeginning Flow run for 'My flow'
Using executor type LocalExecutor
Flow 'My flow': Handling state change from Scheduled to Running
Task 'task_today': Starting task run...
Task 'task_today': Handling state change from Pending to Running
Task 'task_today': Calling task.run() method...
2020-12-11
62f28a1c-822e-47b9-ad04-d97532090daa
Task 'task_today': Handling state change from Running to Success
Task 'task_today': Finished task run for task with final state: 'Success'
Flow run SUCCESS: all reference tasks succeeded
Flow 'My flow': Handling state change from Running to Success
Don't mind the formatting for now.
I would like to see all the message above to be saved in the log fileZanie
liren zhang
12/11/2020, 5:25 PMZanie
liren zhang
12/11/2020, 5:47 PMprefect agent local start --show-flow-logs
and I do see the same log as I saw in the cloud interface. I am thinking if the agent can capture the logs and display them locally in my agent console, I should be able to get it into a log fileZanie
liren zhang
12/11/2020, 5:51 PMZanie