https://prefect.io logo
t

Tony

05/10/2022, 1:36 PM
I maintain a tool to package (set
flow.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?
Copy code
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?
Copy code
from prefect.utilities.storage import extract_flow_from_file

flow = extract_flow_from_file("path")
flow.logger.addHandler()?
. . .
flow.register()
a

Anna Geller

05/10/2022, 1:43 PM
in Prefect 1.0 you can't log from a Flow, you can only log from tasks and ideally by using Prefect logger from context. If you already have some logging library you want to use, you can add that as extra loggers as described here for the same in Prefect 2.0, check this excellent blog post
so to answer your question directly: no, I think the approach you suggested won't work because it would set the logger on a flow-level and this is not possible in Prefect 1.0 allowing to only log from tasks in Prefect 2.0 you will be able to add logs on a flow level
t

Tony

05/10/2022, 1:57 PM
Thanks!
👍 1
c

CA Lee

06/07/2022, 9:08 AM
Hello @Anna Geller, any way to pass print statements from an imported class into a Prefect 2.0 (Orion) task/flow? Example, I have a class that is re-used across multiple Flows.
Copy code
src/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:
Copy code
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):
Copy code
@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()
1
a

Anna Geller

06/07/2022, 11:49 AM
Currently, there is no way of making print statements Prefect logs - I will open an issue for tracking and post here Until then, you can pass the logger object to your custom class to avoid repetition of logger definition
1
71 Views