This message was deleted.
# marvin-ai
s
This message was deleted.
m
thought for 233.9 seconds
Short answer: set the
prefect.task_runs
logger to DEBUG just for the duration of that task run. Here’s a simple pattern you can drop into the task. It temporarily lowers the level, runs your code, then restores the original level:
Copy code
import logging
from prefect import task, get_run_logger

@task
def my_task():
    logger = get_run_logger()

    task_logger = logging.getLogger("prefect.task_runs")
    original_level = task_logger.level
    task_logger.setLevel(logging.DEBUG)
    try:
        logger.debug("Debug message visible only while this task runs")
        # ... your task logic ...
    finally:
        task_logger.setLevel(original_level)
Notes: - This changes the logger level process-wide for
prefect.task_runs
while this task is executing. If you run multiple tasks concurrently in the same process, this may allow DEBUG logs from other tasks, too. - If you need to scope DEBUG strictly to one task run while others remain at INFO, you can add a temporary filter that only lets DEBUG records through for the current task run id:
Copy code
import logging
from prefect import task, get_run_logger
from prefect.context import get_run_context

class OnlyThisTaskDebug(logging.Filter):
    def __init__(self, task_run_id):
        super().__init__()
        self.task_run_id = task_run_id

    def filter(self, record):
        # Allow all INFO+ logs
        if record.levelno >= <http://logging.INFO|logging.INFO>:
            return True
        # Allow DEBUG only for this task run
        return getattr(record, "task_run_id", None) == self.task_run_id

@task
def my_task():
    logger = get_run_logger()
    ctx = get_run_context()

    task_logger = logging.getLogger("prefect.task_runs")
    original_level = task_logger.level
    filt = OnlyThisTaskDebug(ctx.task_run.id)

    # Temporarily allow DEBUG but filter to just this task
    task_logger.addFilter(filt)
    task_logger.setLevel(logging.DEBUG)
    try:
        logger.debug("Only this task's DEBUG logs will appear")
        # ... your task logic ...
    finally:
        task_logger.removeFilter(filt)
        task_logger.setLevel(original_level)
If you’re not running tasks concurrently, the first (simpler) snippet is usually enough. Want me to tailor this to your setup (e.g., concurrent task runner, workers, etc.)?