Tony
05/10/2022, 1:36 PMflow.storage
and flow.run_config
) and registration flows for my enterprise. Recently we wanted to duplicate all Prefect Cloud UI logging to Cloudwatch.
Inside an individual flow I can add this code to get the logs there, but I was wondering if there was a way I could do this through a central utility?
with Flow("My First Flow") as flow:
logger = context.get("logger")
logger.addHandler(
watchtower.CloudWatchLogHandler(
log_group_name="prefect-logs",
)
)
Aka, would something like this work?
from prefect.utilities.storage import extract_flow_from_file
flow = extract_flow_from_file("path")
flow.logger.addHandler()?
. . .
flow.register()
Anna Geller
Tony
05/10/2022, 1:57 PMCA Lee
06/07/2022, 9:08 AMsrc/data.py
class Data:
def __init__(self):
pass
def create(self):
...
print(f"Created.")
def read(self):
...
print(f"Read.")
def update(self):
...
print(f"Updated.")
From a flow.py file:
from src.data import Data
from prefect import flow, task, get_run_logger
@task
def create():
return Data().create()
@task
def read():
return Data().read()
@task
def update():
return Data().update()
@flow
def flow():
created = create()
readed = read(created)
update(readed)
If there is no way to pass print statements into the tasks / flows, I would need to repeat the below code for all flows in Prefect (that were relying on that imported class):
@task
def create():
# These 2 lines need to be written
# fow all tasks / flows
logger = get_run_logger()
<http://logger.info|logger.info>("Created.")
return Data().create()
Anna Geller