feat(django): Add span around Task.enqueue (#5209)

### Description
Add a `queue.submit.django` span when a `Task` in enqueued via Django.

<img width="1352" height="640" alt="Screenshot 2025-12-10 at 13 26 04"
src="https://github.com/user-attachments/assets/5ba32d19-1b73-4a0c-ba59-d55dd7b22f4b"
/>


#### Issues
Closes #5201
Closes PY-2006

#### Reminders
- Please add tests to validate your changes, and lint your code using
`tox -e linters`.
- Add GH Issue ID _&_ Linear ID (if applicable)
- PR title should use [conventional
commit](https://develop.sentry.dev/engineering-practices/commit-messages/#type)
style (`feat:`, `fix:`, `ref:`, `meta:`)
- For external contributors:
[CONTRIBUTING.md](https://github.com/getsentry/sentry-python/blob/master/CONTRIBUTING.md),
[Sentry SDK development docs](https://develop.sentry.dev/sdk/), [Discord
community](https://discord.gg/Ww9hbqr)

Author: sentrivana

sentry_sdk/consts.py

@@ -936,6 +936,7 @@ class OP:
     QUEUE_SUBMIT_RAY = "queue.submit.ray"
     QUEUE_TASK_RAY = "queue.task.ray"
     QUEUE_TASK_DRAMATIQ = "queue.task.dramatiq"
+    QUEUE_SUBMIT_DJANGO = "queue.submit.django"
     SUBPROCESS = "subprocess"
     SUBPROCESS_WAIT = "subprocess.wait"
     SUBPROCESS_COMMUNICATE = "subprocess.communicate"

sentry_sdk/integrations/django/__init__.py

@@ -62,6 +62,7 @@
 )
 from sentry_sdk.integrations.django.middleware import patch_django_middlewares
 from sentry_sdk.integrations.django.signals_handlers import patch_signals
+from sentry_sdk.integrations.django.tasks import patch_tasks
 from sentry_sdk.integrations.django.views import patch_views
 
 if DJANGO_VERSION[:2] > (1, 8):
@@ -271,6 +272,7 @@ def _django_queryset_repr(value, hint):
         patch_views()
         patch_templates()
         patch_signals()
+        patch_tasks()
         add_template_context_repr_sequence()
 
         if patch_caching is not None:

sentry_sdk/integrations/django/tasks.py

@@ -0,0 +1,43 @@
+from functools import wraps
+
+import sentry_sdk
+from sentry_sdk.consts import OP
+from sentry_sdk.tracing import SPANSTATUS
+from sentry_sdk.utils import qualname_from_function
+
+try:
+    # django.tasks were added in Django 6.0
+    from django.tasks.base import Task, TaskResultStatus
+except ImportError:
+    Task = None
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any
+
+
+def patch_tasks():
+    # type: () -> None
+    if Task is None:
+        return
+
+    old_task_enqueue = Task.enqueue
+
+    @wraps(old_task_enqueue)
+    def _sentry_enqueue(self, *args, **kwargs):
+        # type: (Any, Any, Any) -> Any
+        from sentry_sdk.integrations.django import DjangoIntegration
+
+        integration = sentry_sdk.get_client().get_integration(DjangoIntegration)
+        if integration is None:
+            return old_task_enqueue(self, *args, **kwargs)
+
+        name = qualname_from_function(self.func) or "<unknown Django task>"
+
+        with sentry_sdk.start_span(
+            op=OP.QUEUE_SUBMIT_DJANGO, name=name, origin=DjangoIntegration.origin
+        ):
+            return old_task_enqueue(self, *args, **kwargs)
+
+    Task.enqueue = _sentry_enqueue

tests/integrations/django/test_tasks.py

@@ -0,0 +1,187 @@
+import pytest
+
+import sentry_sdk
+from sentry_sdk import start_span
+from sentry_sdk.integrations.django import DjangoIntegration
+from sentry_sdk.consts import OP
+
+
+try:
+    from django.tasks import task
+
+    HAS_DJANGO_TASKS = True
+except ImportError:
+    HAS_DJANGO_TASKS = False
+
+
+@pytest.fixture
+def immediate_backend(settings):
+    """Configure Django to use the immediate task backend for synchronous testing."""
+    settings.TASKS = {
+        "default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}
+    }
+
+
+if HAS_DJANGO_TASKS:
+
+    @task
+    def simple_task():
+        return "result"
+
+    @task
+    def add_numbers(a, b):
+        return a + b
+
+    @task
+    def greet(name, greeting="Hello"):
+        return f"{greeting}, {name}!"
+
+    @task
+    def failing_task():
+        raise ValueError("Task failed!")
+
+    @task
+    def task_one():
+        return 1
+
+    @task
+    def task_two():
+        return 2
+
+
+@pytest.mark.skipif(
+    not HAS_DJANGO_TASKS,
+    reason="Django tasks are only available in Django 6.0+",
+)
+def test_task_span_is_created(sentry_init, capture_events, immediate_backend):
+    """Test that the queue.submit.django span is created when a task is enqueued."""
+    sentry_init(
+        integrations=[DjangoIntegration()],
+        traces_sample_rate=1.0,
+    )
+    events = capture_events()
+
+    with sentry_sdk.start_transaction(name="test_transaction"):
+        simple_task.enqueue()
+
+    (event,) = events
+    assert event["type"] == "transaction"
+
+    queue_submit_spans = [
+        span for span in event["spans"] if span["op"] == OP.QUEUE_SUBMIT_DJANGO
+    ]
+    assert len(queue_submit_spans) == 1
+    assert (
+        queue_submit_spans[0]["description"]
+        == "tests.integrations.django.test_tasks.simple_task"
+    )
+    assert queue_submit_spans[0]["origin"] == "auto.http.django"
+
+
+@pytest.mark.skipif(
+    not HAS_DJANGO_TASKS,
+    reason="Django tasks are only available in Django 6.0+",
+)
+def test_task_enqueue_returns_result(sentry_init, immediate_backend):
+    """Test that the task enqueuing behavior is unchanged from the user perspective."""
+    sentry_init(
+        integrations=[DjangoIntegration()],
+        traces_sample_rate=1.0,
+    )
+
+    result = add_numbers.enqueue(3, 5)
+
+    assert result is not None
+    assert result.return_value == 8
+
+
+@pytest.mark.skipif(
+    not HAS_DJANGO_TASKS,
+    reason="Django tasks are only available in Django 6.0+",
+)
+def test_task_enqueue_with_kwargs(sentry_init, immediate_backend, capture_events):
+    """Test that task enqueuing works correctly with keyword arguments."""
+    sentry_init(
+        integrations=[DjangoIntegration()],
+        traces_sample_rate=1.0,
+    )
+    events = capture_events()
+
+    with sentry_sdk.start_transaction(name="test_transaction"):
+        result = greet.enqueue(name="World", greeting="Hi")
+
+    assert result.return_value == "Hi, World!"
+
+    (event,) = events
+    queue_submit_spans = [
+        span for span in event["spans"] if span["op"] == OP.QUEUE_SUBMIT_DJANGO
+    ]
+    assert len(queue_submit_spans) == 1
+    assert (
+        queue_submit_spans[0]["description"]
+        == "tests.integrations.django.test_tasks.greet"
+    )
+
+
+@pytest.mark.skipif(
+    not HAS_DJANGO_TASKS,
+    reason="Django tasks are only available in Django 6.0+",
+)
+def test_task_error_reporting(sentry_init, immediate_backend, capture_events):
+    """Test that errors in tasks are correctly reported and don't break the span."""
+    sentry_init(
+        integrations=[DjangoIntegration()],
+        traces_sample_rate=1.0,
+    )
+    events = capture_events()
+
+    with sentry_sdk.start_transaction(name="test_transaction"):
+        result = failing_task.enqueue()
+
+    with pytest.raises(ValueError, match="Task failed"):
+        _ = result.return_value
+
+    assert len(events) == 2
+    transaction_event = events[-1]
+    assert transaction_event["type"] == "transaction"
+
+    queue_submit_spans = [
+        span
+        for span in transaction_event["spans"]
+        if span["op"] == OP.QUEUE_SUBMIT_DJANGO
+    ]
+    assert len(queue_submit_spans) == 1
+    assert (
+        queue_submit_spans[0]["description"]
+        == "tests.integrations.django.test_tasks.failing_task"
+    )
+
+
+@pytest.mark.skipif(
+    not HAS_DJANGO_TASKS,
+    reason="Django tasks are only available in Django 6.0+",
+)
+def test_multiple_task_enqueues_create_multiple_spans(
+    sentry_init, capture_events, immediate_backend
+):
+    """Test that enqueueing multiple tasks creates multiple spans."""
+    sentry_init(
+        integrations=[DjangoIntegration()],
+        traces_sample_rate=1.0,
+    )
+    events = capture_events()
+
+    with sentry_sdk.start_transaction(name="test_transaction"):
+        task_one.enqueue()
+        task_two.enqueue()
+        task_one.enqueue()
+
+    (event,) = events
+    queue_submit_spans = [
+        span for span in event["spans"] if span["op"] == OP.QUEUE_SUBMIT_DJANGO
+    ]
+    assert len(queue_submit_spans) == 3
+
+    span_names = [span["description"] for span in queue_submit_spans]
+    assert span_names.count("tests.integrations.django.test_tasks.task_one") == 2
+    assert span_names.count("tests.integrations.django.test_tasks.task_two") == 1