Hey <@ULVA73B9P> , i am trying to test the tasks i...
# marvin-ai
t
Hey @Marvin , i am trying to test the tasks in my codebase but i get a logging warning whenever i run pytest. I use get_run_logger in every task so probably this causes the issue. How can i omit this warning?
Copy code
--- Logging error ---
site-packages/prefect/logging/handlers.py", line 355, in emit
...
Message: 'Still processing items: 3 items remaining...'
Arguments: ()
m
thought for 217.8 seconds
You’re hitting Prefect’s API log worker shutting down while there are still log records in its queue. During pytest teardown it logs “Still processing items: N items remaining…” while draining. It’s harmless, but noisy—especially if you call
get_run_logger()
in many tasks. Two easy ways to silence it in Prefect 3.x: Option A — Disable API logging during tests (recommended) - Keeps logs in local stdout/stderr; avoids the background API log worker entirely.
Copy code
# conftest.py
import pytest
from prefect.settings import (
    temporary_settings,
    PREFECT_LOGGING_TO_API_ENABLED,
    PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW,
)

@pytest.fixture(autouse=True)
def disable_prefect_api_logging():
    # Prevent the API log handler from starting in tests
    with temporary_settings(
        updates={
            PREFECT_LOGGING_TO_API_ENABLED: False,
            # Avoid warnings if get_run_logger is used outside a run context
            PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW: "ignore",
        }
    ):
        yield
If you prefer environment variables, you can export these before running pytest:
Copy code
PREFECT_LOGGING_TO_API_ENABLED=false
PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW=ignore
Option B — Keep API logging, but flush/drain the worker at teardown - If you need API logging in tests, make sure the queue is drained cleanly.
Copy code
# conftest.py
import asyncio
import pytest
from prefect.logging.handlers import APILogHandler, APILogWorker

@pytest.fixture(scope="session", autouse=True)
def drain_prefect_log_workers_on_exit():
    yield
    # Ensure everything is flushed before pytest exits
    asyncio.run(APILogHandler.aflush())
    asyncio.run(APILogWorker.drain_all())
Tips - If you directly call task functions (outside a flow context), also set
PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW=ignore
to suppress extra warnings. - To capture run logs in tests, prefer
LogEavesdropper
or add the caplog handler to
prefect.flow_runs
/
prefect.task_runs
instead of relying on the API handler. If you can share your Prefect version and whether you want logs shipped to the Prefect API during tests, I can tailor the fixture exactly to your setup.
t
@Marvin i am using Prefect version 3.5.0 i dont want logs shipped to the prefect api during tests
m
thought for 38.0 seconds
Got it — Prefect 3.5.0 and you don’t want logs shipped to the Prefect API during tests. That warning is from the API log worker shutting down with items still in its queue. The clean fix is to disable API logging in tests and (for safety) drain any worker that a test might have started. Pick one of these: Option 1: Set env vars before pytest (simplest)
Copy code
export PREFECT_LOGGING_TO_API_ENABLED=false
export PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW=ignore
pytest
Option 2: Conftest fixtures (project-local and robust)
Copy code
# conftest.py
import asyncio
import pytest
from prefect.settings import (
    temporary_settings,
    PREFECT_LOGGING_TO_API_ENABLED,
    PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW,
)
from prefect.logging.handlers import APILogHandler, APILogWorker

@pytest.fixture(autouse=True)
def disable_prefect_api_logging():
    # Prevent the API log handler from running during tests
    with temporary_settings(
        updates={
            PREFECT_LOGGING_TO_API_ENABLED: False,
            # Suppress warnings if get_run_logger is used without a run context
            PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW: "ignore",
        }
    ):
        yield

@pytest.fixture(scope="session", autouse=True)
def drain_prefect_log_workers_on_exit():
    # Ensure any worker started by a test is fully drained to avoid warnings
    yield
    asyncio.run(APILogHandler.aflush())
    asyncio.run(APILogWorker.drain_all())
Notes - This will keep your
get_run_logger()
logs local (stdout/stderr) and silence the “Still processing items…” warning. - If you ever need API logging for a specific test, wrap that test body with:
Copy code
from prefect.settings import temporary_settings, PREFECT_LOGGING_TO_API_ENABLED

  with temporary_settings(updates={PREFECT_LOGGING_TO_API_ENABLED: True}):
      # run the test code that should ship logs