Ritesh
06/28/2025, 7:31 PMMarvin
06/28/2025, 7:31 PMMarvin
06/28/2025, 7:32 PM<http://logger.info|logger.info>()
statements won't automatically show up in the Prefect UI. To see your logs in the Prefect UI, you'll need to make a small modification to your code to use Prefect's built-in logging system.
Here's how to properly set up logging in your Prefect flows:
python
from prefect import flow, task
from prefect.logging import get_run_logger
@task
def my_task():
logger = get_run_logger() # Get Prefect's logger
<http://logger.info|logger.info>("This will show up in the Prefect UI!")
@flow
def my_flow():
logger = get_run_logger() # Get Prefect's logger
<http://logger.info|logger.info>("This flow log will appear in the UI")
my_task()
If you want to keep your existing logger.info statements and also see them in the Prefect UI, you have two options:
1. Replace your existing logger with Prefect's run logger using get_run_logger()
2. Configure your Python logger to also send logs to Prefect's logging system (this is more advanced and generally not recommended)
Additionally, if you're using print statements and want those to show up in the UI, you can enable that by setting log_prints=True
in your flow or task decorators:
python
@flow(log_prints=True)
def my_flow():
print("This will also show up in the UI!")
For more details on logging configuration and best practices, you can check out the Prefect logging documentation.Ritesh
06/28/2025, 7:35 PMMarvin
06/28/2025, 7:36 PMMarvin
06/28/2025, 7:36 PMget_run_logger()
must be called within a flow or task context to work properly
2. It needs to be called each time you want to log within a flow/task to get the correct runtime context
Here's what NOT to do:
python
# This won't work as expected
from prefect.logging import get_run_logger as logger
@flow
def my_flow():
<http://logger.info|logger.info>("This won't show up properly in the UI") # Wrong!
Here's the correct way to use it:
python
from prefect import flow, task
from prefect.logging import get_run_logger
@task
def my_task():
logger = get_run_logger() # Get logger inside the task
<http://logger.info|logger.info>("Task log - will show in UI!")
@flow
def my_flow():
logger = get_run_logger() # Get logger inside the flow
<http://logger.info|logger.info>("Flow log - will show in UI!")
my_task()
The key difference is that get_run_logger()
needs to be called within each flow or task where you want to log messages. This ensures the logs are properly associated with the correct flow or task run in the Prefect UI.
If you have many logging statements and want to avoid repeatedly calling get_run_logger()
, you could create a helper function within your flow or task:
python
@flow
def my_flow():
def get_logger():
return get_run_logger()
logger = get_logger()
<http://logger.info|logger.info>("First message")
# ... do some work ...
<http://logger.info|logger.info>("Another message")
Remember: The logger needs the runtime context of a flow or task to properly send logs to the Prefect UI, which is why it can't be initialized at the module level.Ritesh
06/28/2025, 7:38 PMMarvin
06/28/2025, 7:38 PMMarvin
06/28/2025, 7:39 PMlogger = get_run_logger()
globally will cause problems because:
1. It will raise a MissingContextError
when your module is imported because there's no flow or task context available at that time
2. Even if it didn't raise an error, the logs wouldn't be properly associated with your flow and task runs in the UI
Here's what happens:
python
# This will fail with MissingContextError
from prefect.logging import get_run_logger
logger = get_run_logger() # ❌ ERROR: No runtime context available!
@flow
def my_flow():
<http://logger.info|logger.info>("This wouldn't work anyway")
Instead, you should always get the logger within your flows and tasks. Here are a few patterns you can use to make this more convenient:
1. Basic Pattern (most straightforward):
python
@flow
def my_flow():
logger = get_run_logger()
<http://logger.info|logger.info>("This works!")
2. Helper Function Pattern (if you have many logging statements in one flow):
python
@flow
def my_flow():
def get_logger():
return get_run_logger()
logger = get_logger()
<http://logger.info|logger.info>("Message 1")
<http://logger.info|logger.info>("Message 2")
3. Class-based Pattern (if you're using classes):
python
class MyWorkflow:
@flow
def run(self):
self.logger = get_run_logger()
<http://self.logger.info|self.logger.info>("Now we can use self.logger in this flow")
If you want to maintain both regular Python logging and Prefect logging, you might want to set up a proper Python logging configuration that handles both cases, but that's a more advanced setup. For most use cases, simply getting the logger within each flow or task is the recommended approach.
Remember: The key concept is that Prefect's logger needs to know which flow or task run it's logging for, which is only possible when called within that flow or task's context.Ritesh
06/28/2025, 7:44 PMMarvin
06/28/2025, 7:44 PMMarvin
06/28/2025, 7:46 PMpython
from prefect.client import get_client
from prefect import get_client
async def get_flow_run_logs(flow_run_id):
async with get_client() as client:
logs = await client.read_logs(flow_run_id=flow_run_id)
return logs
2. Using the Prefect CLI:
bash
prefect flow-run logs <flow-run-id>
3. Using Direct Database Access (if you're using a local SQLite database or have access to your PostgreSQL database):
- Logs are stored in the log
table of your Prefect database
- You can query them directly using SQL if you have database access
4. Using the REST API:
python
import requests
def get_logs(api_url, flow_run_id, api_key):
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(
f"{api_url}/flow-runs/{flow_run_id}/logs",
headers=headers
)
return response.json()
5. Save Logs During Flow Execution:
If you want to proactively save logs for future reference, you can modify your flows to write logs to a file:
python
from prefect import flow
from prefect.logging import get_run_logger
import logging
# Set up file handler
file_handler = logging.FileHandler('my_flow_logs.log')
file_handler.setLevel(<http://logging.INFO|logging.INFO>)
@flow
def my_flow():
logger = get_run_logger()
# Add the file handler to the logger
logger.addHandler(file_handler)
<http://logger.info|logger.info>("This will be saved to both Prefect UI and file")
For bulk export of historical logs, the API method is probably your best bet as it allows you to programmatically fetch logs for multiple flow runs. Here's a more complete example:
python
from prefect import get_client
import asyncio
import json
async def export_logs_for_timeframe(start_time, end_time):
async with get_client() as client:
# Get flow runs in the time period
flow_runs = await client.read_flow_runs(
start_time=start_time,
end_time=end_time
)
all_logs = []
for flow_run in flow_runs:
logs = await client.read_logs(flow_run_id=flow_run.id)
all_logs.extend(logs)
# Save to file
with open('prefect_logs_export.json', 'w') as f:
json.dump(all_logs, f, indent=2)
# Run the export
asyncio.run(export_logs_for_timeframe(
start_time="2023-01-01T00:00:00Z",
end_time="2023-12-31T23:59:59Z"
))
Remember that the availability of logs depends on your Prefect server's retention settings and the database backend you're using. If you need to maintain logs for extended periods, consider implementing the file-based logging approach alongside the Prefect UI logging.