Calling a Django Management Command from a Celery Task and Redirecting Logs

I'm trying to call a Management Command with call_command within a Celery task. The management command writes logs in different ways: through print statements, self.stdout.write, or using the logger defined at the script's module level.

I want to capture all these log messages in the Celery logger to make them visible in Datadog. Is this possible?

I tried the worker_redirect_stdouts_level option, but it impacts the global Celery logging configuration.

You can use call_command() with StringIO to capture stdout within a Celery task and redirect it to the Celery logger

from celery import shared_task
from django.core.management import call_command
import io
import logging

logger = logging.getLogger(__name__)

@shared_task
def task_that_runs_your_management_command():
    output = io.StringIO()  # To capture stdout
    
    try:
        call_command("your_command", stdout=output, stderr=output)
    except Exception as e:
        logger.exception("Error while executing management command")

    # Log the captured output
    command_output = output.getvalue()
    if command_output:
        logger.info("Management Command Output:\n%s", command_output)
Back to Top