diff --git a/sentry_sdk/integrations/_asgi_common.py b/sentry_sdk/integrations/_asgi_common.py
index dbf02decdc..e668a37aeb 100644
--- a/sentry_sdk/integrations/_asgi_common.py
+++ b/sentry_sdk/integrations/_asgi_common.py
@@ -4,7 +4,7 @@
 
 import sentry_sdk
 from sentry_sdk.data_collection import _apply_data_collection_filtering_to_query_string
-from sentry_sdk.integrations._wsgi_common import _filter_headers
+from sentry_sdk.integrations._wsgi_common import _filter_headers_legacy
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.utils import has_data_collection_enabled
 
@@ -121,7 +121,7 @@ def _get_request_data(
 
         headers = _get_headers(asgi_scope)
 
-        request_data["headers"] = _filter_headers(
+        request_data["headers"] = _filter_headers_legacy(
             headers,
             use_annotated_value=False,
         )
@@ -175,7 +175,7 @@ def _get_request_attributes(
 
         headers = _get_headers(asgi_scope)
 
-        filtered_headers = _filter_headers(headers, use_annotated_value=False)
+        filtered_headers = _filter_headers_legacy(headers, use_annotated_value=False)
         for header, value in filtered_headers.items():
             attributes[f"http.request.header.{header.lower()}"] = value
 
diff --git a/sentry_sdk/integrations/_wsgi_common.py b/sentry_sdk/integrations/_wsgi_common.py
index eaf764fb4b..2537946d0b 100644
--- a/sentry_sdk/integrations/_wsgi_common.py
+++ b/sentry_sdk/integrations/_wsgi_common.py
@@ -23,6 +23,18 @@
     from sentry_sdk._types import Event
 
 
+DEFAULT_HTTP_METHODS_TO_CAPTURE = (
+    "CONNECT",
+    "DELETE",
+    "GET",
+    # "HEAD",  # do not capture HEAD requests by default
+    # "OPTIONS",  # do not capture OPTIONS requests by default
+    "PATCH",
+    "POST",
+    "PUT",
+    "TRACE",
+)
+
 SENSITIVE_ENV_KEYS = (
     "REMOTE_ADDR",
     "HTTP_X_FORWARDED_FOR",
@@ -39,18 +51,6 @@
     x[len("HTTP_") :] for x in SENSITIVE_ENV_KEYS if x.startswith("HTTP_")
 )
 
-DEFAULT_HTTP_METHODS_TO_CAPTURE = (
-    "CONNECT",
-    "DELETE",
-    "GET",
-    # "HEAD",  # do not capture HEAD requests by default
-    # "OPTIONS",  # do not capture OPTIONS requests by default
-    "PATCH",
-    "POST",
-    "PUT",
-    "TRACE",
-)
-
 
 def request_body_within_bounds(
     client: "Optional[sentry_sdk.client.BaseClient]", content_length: int
@@ -89,23 +89,15 @@ def extract_into_event(self, event: "Event") -> None:
         content_length = self.content_length()
         request_info = event.get("request", {})
 
-        # Prior to data collection being implemented we unconditionally attached
-        # the request body, which is why we default to True here.
-        attach_request_body = True
-
-        if has_data_collection_enabled(client.options):
-            cookies = _apply_key_value_collection_filtering(
-                items=dict(self.cookies()),
-                behaviour=client.options["data_collection"]["cookies"],
-            )
-            if cookies:
-                request_info["cookies"] = cookies
+        data_collection = client.options["data_collection"]
+        cookies = _apply_key_value_collection_filtering(
+            items=dict(self.cookies()),
+            behaviour=data_collection["cookies"],
+        )
+        if cookies:
+            request_info["cookies"] = cookies
 
-            attach_request_body = (
-                "incoming_request" in client.options["data_collection"]["http_bodies"]
-            )
-        elif should_send_default_pii():
-            request_info["cookies"] = dict(self.cookies())
+        attach_request_body = "incoming_request" in data_collection["http_bodies"]
 
         if attach_request_body:
             if not request_body_within_bounds(client, content_length):
@@ -210,6 +202,63 @@ def env(self) -> "Dict[str, Any]":
         raise NotImplementedError()
 
 
+class LegacyRequestExtractor(RequestExtractor):
+    def extract_into_event(self, event: "Event") -> None:
+        client = sentry_sdk.get_client()
+        if not client.is_active():
+            return
+
+        data: "Optional[Union[AnnotatedValue, Dict[str, Any]]]" = None
+        content_length = self.content_length()
+        request_info = event.get("request", {})
+
+        # Prior to data collection being implemented we unconditionally attached
+        # the request body, which is why we default to True here.
+        attach_request_body = True
+        if has_data_collection_enabled(client.options):
+            cookies = _apply_key_value_collection_filtering(
+                items=dict(self.cookies()),
+                behaviour=client.options["data_collection"]["cookies"],
+            )
+            if cookies:
+                request_info["cookies"] = cookies
+
+            attach_request_body = (
+                "incoming_request" in client.options["data_collection"]["http_bodies"]
+            )
+        elif should_send_default_pii():
+            request_info["cookies"] = dict(self.cookies())
+
+        if attach_request_body:
+            if not request_body_within_bounds(client, content_length):
+                data = AnnotatedValue.removed_because_over_size_limit()
+            else:
+                # First read the raw body data
+                # It is important to read this first because if it is Django
+                # it will cache the body and then we can read the cached version
+                # again in parsed_body() (or json() or wherever).
+                raw_data = None
+                try:
+                    raw_data = self.raw_data()
+                except _RAW_DATA_EXCEPTIONS:
+                    # If DjangoRestFramework is used it already read the body for us
+                    # so reading it here will fail. We can ignore this.
+                    pass
+
+                parsed_body = self.parsed_body()
+                if parsed_body is not None:
+                    data = parsed_body
+                elif raw_data:
+                    data = AnnotatedValue.removed_because_raw_data()
+                else:
+                    data = None
+
+        if data is not None:
+            request_info["data"] = data
+
+        event["request"] = deepcopy(request_info)
+
+
 def _is_json_content_type(ct: "Optional[str]") -> bool:
     mt = (ct or "").split(";", 1)[0]
     return (
@@ -222,12 +271,29 @@ def _is_json_content_type(ct: "Optional[str]") -> bool:
 def _filter_headers(
     headers: "Mapping[str, str]",
     use_annotated_value: bool = True,
+) -> "Mapping[str, Union[AnnotatedValue, str]]":
+    client = sentry_sdk.get_client()
+
+    filtered = _apply_key_value_collection_filtering(
+        items=headers,
+        behaviour=client.options["data_collection"]["http_headers"]["request"],
+    )
+
+    for key in filtered:
+        if isinstance(key, str) and key.lower() in ("cookie", "set-cookie"):
+            filtered[key] = SENSITIVE_DATA_SUBSTITUTE
+
+    return filtered
+
+
+def _filter_headers_legacy(
+    headers: "Mapping[str, str]",
+    use_annotated_value: bool = True,
 ) -> "Mapping[str, Union[AnnotatedValue, str]]":
     client_options = sentry_sdk.get_client().options
 
     if has_data_collection_enabled(client_options):
         data_collection_configuration = client_options["data_collection"]
-
         filtered = _apply_key_value_collection_filtering(
             items=headers,
             behaviour=data_collection_configuration["http_headers"]["request"],
@@ -238,6 +304,7 @@ def _filter_headers(
                 filtered[key] = SENSITIVE_DATA_SUBSTITUTE
 
         return filtered
+
     else:
         if should_send_default_pii():
             return headers
diff --git a/sentry_sdk/integrations/django/__init__.py b/sentry_sdk/integrations/django/__init__.py
index c362277038..b8aebcba64 100644
--- a/sentry_sdk/integrations/django/__init__.py
+++ b/sentry_sdk/integrations/django/__init__.py
@@ -15,7 +15,7 @@
 )
 from sentry_sdk.integrations._wsgi_common import (
     DEFAULT_HTTP_METHODS_TO_CAPTURE,
-    RequestExtractor,
+    LegacyRequestExtractor,
 )
 from sentry_sdk.integrations.logging import ignore_logger_for_events
 from sentry_sdk.integrations.wsgi import SentryWsgiMiddleware
@@ -703,7 +703,7 @@ def sentry_patched_response_for_exception(
     exception_handler.response_for_exception = sentry_patched_response_for_exception
 
 
-class DjangoRequestExtractor(RequestExtractor):
+class DjangoRequestExtractor(LegacyRequestExtractor):
     def __init__(self, request: "Union[WSGIRequest, ASGIRequest]") -> None:
         try:
             drf_request = request._sentry_drf_request_backref()
@@ -747,7 +747,7 @@ def parsed_body(self) -> "Optional[Dict[str, Any]]":
         try:
             return self.request.data
         except Exception:
-            return RequestExtractor.parsed_body(self)
+            return LegacyRequestExtractor.parsed_body(self)
 
 
 def _set_user_info(request: "WSGIRequest", event: "Event") -> None:
diff --git a/sentry_sdk/integrations/flask.py b/sentry_sdk/integrations/flask.py
index bff656bbb8..2285ca73bc 100644
--- a/sentry_sdk/integrations/flask.py
+++ b/sentry_sdk/integrations/flask.py
@@ -5,7 +5,7 @@
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.integrations._wsgi_common import (
     DEFAULT_HTTP_METHODS_TO_CAPTURE,
-    RequestExtractor,
+    LegacyRequestExtractor,
 )
 from sentry_sdk.integrations.wsgi import SentryWsgiMiddleware
 from sentry_sdk.scope import should_send_default_pii
@@ -185,7 +185,7 @@ def _request_started(app: "Flask", **kwargs: "Any") -> None:
     scope.add_event_processor(evt_processor)
 
 
-class FlaskRequestExtractor(RequestExtractor):
+class FlaskRequestExtractor(LegacyRequestExtractor):
     def env(self) -> "Dict[str, str]":
         return self.request.environ
 
diff --git a/sentry_sdk/integrations/gcp.py b/sentry_sdk/integrations/gcp.py
index 1b53889a25..a0769a91b0 100644
--- a/sentry_sdk/integrations/gcp.py
+++ b/sentry_sdk/integrations/gcp.py
@@ -9,7 +9,7 @@
 from sentry_sdk.consts import OP
 from sentry_sdk.data_collection import _apply_data_collection_filtering_to_query_string
 from sentry_sdk.integrations import Integration
-from sentry_sdk.integrations._wsgi_common import _filter_headers
+from sentry_sdk.integrations._wsgi_common import _filter_headers_legacy
 from sentry_sdk.integrations.cloud_resource_context import CLOUD_PROVIDER
 from sentry_sdk.scope import Scope, should_send_default_pii
 from sentry_sdk.traces import SegmentNameSource
@@ -86,7 +86,7 @@ def sentry_func(
             header_attributes: "dict[str, Any]" = {}
             if hasattr(gcp_event, "headers"):
                 headers = gcp_event.headers
-                for header, header_value in _filter_headers(
+                for header, header_value in _filter_headers_legacy(
                     headers, use_annotated_value=False
                 ).items():
                     header_attributes[f"http.request.header.{header.lower()}"] = (
@@ -244,7 +244,7 @@ def event_processor(event: "Event", hint: "Hint") -> "Optional[Event]":
                 request["query_string"] = query_string
 
         if hasattr(gcp_event, "headers"):
-            request["headers"] = _filter_headers(gcp_event.headers)
+            request["headers"] = _filter_headers_legacy(gcp_event.headers)
 
         if hasattr(gcp_event, "data"):
             if has_data_collection_enabled(client_options):
diff --git a/sentry_sdk/integrations/quart.py b/sentry_sdk/integrations/quart.py
index 1343867fc3..e2edb69db3 100644
--- a/sentry_sdk/integrations/quart.py
+++ b/sentry_sdk/integrations/quart.py
@@ -8,7 +8,7 @@
 from sentry_sdk.consts import SPANDATA
 from sentry_sdk.data_collection import _apply_data_collection_filtering_to_query_string
 from sentry_sdk.integrations import DidNotEnable, Integration
-from sentry_sdk.integrations._wsgi_common import _filter_headers
+from sentry_sdk.integrations._wsgi_common import _filter_headers_legacy
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.traces import SOURCE_FOR_STYLE as SEGMENT_SOURCE_FOR_STYLE
@@ -187,7 +187,7 @@ async def _request_websocket_started(app: "Quart", **kwargs: "Any") -> None:
         segment.set_attribute("http.request.method", request_websocket.method)
         header_attributes: "dict[str, Any]" = {}
 
-        for header, header_value in _filter_headers(
+        for header, header_value in _filter_headers_legacy(
             dict(request_websocket.headers), use_annotated_value=False
         ).items():
             header_attributes[f"http.request.header.{header.lower()}"] = header_value
@@ -282,7 +282,7 @@ def inner(event: "Event", hint: "dict[str, Any]") -> "Event":
             request_info["url"] = request.url
             request_info["query_string"] = request.query_string
             request_info["method"] = request.method
-            request_info["headers"] = _filter_headers(dict(request.headers))
+            request_info["headers"] = _filter_headers_legacy(dict(request.headers))
 
             client_options = sentry_sdk.get_client().options
             if has_data_collection_enabled(client_options):
diff --git a/sentry_sdk/integrations/sanic.py b/sentry_sdk/integrations/sanic.py
index be536b2a4f..486338cbde 100644
--- a/sentry_sdk/integrations/sanic.py
+++ b/sentry_sdk/integrations/sanic.py
@@ -10,7 +10,10 @@
     _apply_data_collection_filtering_to_query_string,
 )
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
-from sentry_sdk.integrations._wsgi_common import RequestExtractor, _filter_headers
+from sentry_sdk.integrations._wsgi_common import (
+    LegacyRequestExtractor,
+    _filter_headers_legacy,
+)
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.traces import SegmentNameSource
 from sentry_sdk.utils import (
@@ -64,7 +67,7 @@ def setup_once() -> None:
         _setup_sanic()
 
 
-class SanicRequestExtractor(RequestExtractor):
+class SanicRequestExtractor(LegacyRequestExtractor):
     def content_length(self) -> int:
         if self.request.body is None:
             return 0
@@ -251,7 +254,7 @@ def _get_request_attributes(request: "Request") -> "Dict[str, Any]":
     if request.method:
         attributes[SPANDATA.HTTP_REQUEST_METHOD] = request.method.upper()
 
-    headers = _filter_headers(dict(request.headers), use_annotated_value=False)
+    headers = _filter_headers_legacy(dict(request.headers), use_annotated_value=False)
     for header, value in headers.items():
         attributes[f"{SPANDATA.HTTP_REQUEST_HEADER}.{header.lower()}"] = value
 
@@ -340,7 +343,7 @@ def sanic_processor(event: "Event", hint: "Optional[Hint]") -> "Optional[Event]"
                 or client_options["data_collection"]["user_info"]
             ):
                 request_info["env"] = {"REMOTE_ADDR": request.remote_addr}
-            request_info["headers"] = _filter_headers(dict(request.headers))
+            request_info["headers"] = _filter_headers_legacy(dict(request.headers))
 
         return event
 
diff --git a/sentry_sdk/integrations/tornado.py b/sentry_sdk/integrations/tornado.py
index 0dba1f8e46..e682a200fe 100644
--- a/sentry_sdk/integrations/tornado.py
+++ b/sentry_sdk/integrations/tornado.py
@@ -6,8 +6,8 @@
 from sentry_sdk.data_collection import _apply_data_collection_filtering_to_query_string
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.integrations._wsgi_common import (
-    RequestExtractor,
-    _filter_headers,
+    LegacyRequestExtractor,
+    _filter_headers_legacy,
     _is_json_content_type,
     request_body_within_bounds,
 )
@@ -147,7 +147,7 @@ def _get_request_attributes(request: "Any") -> "Dict[str, Any]":
     if request.method:
         attributes[SPANDATA.HTTP_REQUEST_METHOD] = request.method.upper()
 
-    headers = _filter_headers(dict(request.headers), use_annotated_value=False)
+    headers = _filter_headers_legacy(dict(request.headers), use_annotated_value=False)
     for header, value in headers.items():
         attributes[f"{SPANDATA.HTTP_REQUEST_HEADER}.{header.lower()}"] = value
 
@@ -280,7 +280,7 @@ def tornado_processor(event: "Event", hint: "dict[str, Any]") -> "Event":
                 or client_options["data_collection"]["user_info"]
             ):
                 request_info["env"] = {"REMOTE_ADDR": request.remote_ip}
-            request_info["headers"] = _filter_headers(dict(request.headers))
+            request_info["headers"] = _filter_headers_legacy(dict(request.headers))
 
         if has_data_collection_enabled(client_options):
             if client_options["data_collection"]["user_info"]:
@@ -305,7 +305,7 @@ def tornado_processor(event: "Event", hint: "dict[str, Any]") -> "Event":
     return tornado_processor
 
 
-class TornadoRequestExtractor(RequestExtractor):
+class TornadoRequestExtractor(LegacyRequestExtractor):
     def content_length(self) -> int:
         if self.request.body is None:
             return 0
diff --git a/sentry_sdk/integrations/wsgi.py b/sentry_sdk/integrations/wsgi.py
index a4df7d1dc2..259cbb6165 100644
--- a/sentry_sdk/integrations/wsgi.py
+++ b/sentry_sdk/integrations/wsgi.py
@@ -12,13 +12,12 @@
     DEFAULT_HTTP_METHODS_TO_CAPTURE,
     _filter_headers,
 )
-from sentry_sdk.scope import Scope, should_send_default_pii, use_isolation_scope
+from sentry_sdk.scope import Scope, use_isolation_scope
 from sentry_sdk.sessions import track_session
 from sentry_sdk.traces import SegmentNameSource
 from sentry_sdk.utils import (
     capture_internal_exceptions,
     event_from_exception,
-    has_data_collection_enabled,
     reraise,
 )
 
@@ -132,14 +131,7 @@ def __call__(
                         sentry_sdk.continue_trace(dict(_get_headers(environ)))
                         Scope.set_custom_sampling_context({"wsgi_environ": environ})
 
-                        if has_data_collection_enabled(client.options):
-                            if client.options["data_collection"]["user_info"]:
-                                client_ip = get_client_ip(environ)
-                                if client_ip:
-                                    scope.set_attribute(
-                                        SPANDATA.USER_IP_ADDRESS, client_ip
-                                    )
-                        elif should_send_default_pii():
+                        if client.options["data_collection"]["user_info"]:
                             client_ip = get_client_ip(environ)
                             if client_ip:
                                 scope.set_attribute(SPANDATA.USER_IP_ADDRESS, client_ip)
@@ -224,17 +216,14 @@ def _sentry_start_response(
 def _get_environ(environ: "Dict[str, str]") -> "Iterator[Tuple[str, str]]":
     """
     Returns our explicitly included environment variables we want to
-    capture (server name, port and remote addr if pii is enabled).
+    capture (server name, port and remote addr if `user_info` is enabled).
     """
     keys = ["SERVER_NAME", "SERVER_PORT"]
     client_options = sentry_sdk.get_client().options
 
     # make debugging of proxy setup easier. Proxy headers are
     # in headers.
-    if has_data_collection_enabled(client_options):
-        if client_options["data_collection"]["user_info"]:
-            keys += ["REMOTE_ADDR"]
-    elif should_send_default_pii():
+    if client_options["data_collection"]["user_info"]:
         keys += ["REMOTE_ADDR"]
 
     for key in keys:
@@ -353,12 +342,7 @@ def event_processor(event: "Event", hint: "Dict[str, Any]") -> "Event":
             # if the code below fails halfway through we at least have some data
             request_info = event.setdefault("request", {})
 
-            if has_data_collection_enabled(client_options):
-                if client_options["data_collection"]["user_info"]:
-                    user_info = event.setdefault("user", {})
-                    if client_ip:
-                        user_info.setdefault("ip_address", client_ip)
-            elif should_send_default_pii():
+            if client_options["data_collection"]["user_info"]:
                 user_info = event.setdefault("user", {})
                 if client_ip:
                     user_info.setdefault("ip_address", client_ip)
@@ -368,17 +352,13 @@ def event_processor(event: "Event", hint: "Dict[str, Any]") -> "Event":
             request_info["env"] = env
             request_info["headers"] = headers
 
-            if has_data_collection_enabled(client_options):
-                if query_string:
-                    filtered_qs = _apply_data_collection_filtering_to_query_string(
-                        query_string=query_string,
-                        behaviour=client_options["data_collection"]["url_query_params"],
-                    )
-                    if filtered_qs:
-                        request_info["query_string"] = filtered_qs
-            else:
-                # This was not originally gated so if data collection is not enabled, leave as-is.
-                request_info["query_string"] = query_string
+            if query_string:
+                filtered_qs = _apply_data_collection_filtering_to_query_string(
+                    query_string=query_string,
+                    behaviour=client_options["data_collection"]["url_query_params"],
+                )
+                if filtered_qs:
+                    request_info["query_string"] = filtered_qs
 
         return event
 
@@ -419,47 +399,28 @@ def _get_request_attributes(
 
     client_options = sentry_sdk.get_client().options
 
-    if has_data_collection_enabled(client_options):
-        query_string = environ.get("QUERY_STRING")
-        filtered_qs = None
-        if query_string:
-            filtered_qs = _apply_data_collection_filtering_to_query_string(
-                query_string=query_string,
-                behaviour=client_options["data_collection"]["url_query_params"],
-            )
-
-            if filtered_qs:
-                attributes["http.query"] = filtered_qs
+    query_string = environ.get("QUERY_STRING")
+    filtered_qs = None
+    if query_string:
+        filtered_qs = _apply_data_collection_filtering_to_query_string(
+            query_string=query_string,
+            behaviour=client_options["data_collection"]["url_query_params"],
+        )
 
-        path = environ.get("PATH_INFO", "")
-        if path:
-            attributes["url.path"] = path
+        if filtered_qs:
+            attributes["http.query"] = filtered_qs
 
-        attributes["url.full"] = get_request_url(environ, use_x_forwarded_for)
-        if filtered_qs is not None:
-            attributes["url.full"] += f"?{filtered_qs}"
+    path = environ.get("PATH_INFO", "")
+    if path:
+        attributes["url.path"] = path
 
-        if client_options["data_collection"]["user_info"]:
-            client_ip = get_client_ip(environ)
-            if client_ip:
-                attributes["client.address"] = client_ip
+    attributes["url.full"] = get_request_url(environ, use_x_forwarded_for)
+    if filtered_qs is not None:
+        attributes["url.full"] += f"?{filtered_qs}"
 
-    elif should_send_default_pii():
+    if client_options["data_collection"]["user_info"]:
         client_ip = get_client_ip(environ)
         if client_ip:
             attributes["client.address"] = client_ip
 
-        query_string = environ.get("QUERY_STRING")
-        if query_string:
-            attributes["http.query"] = query_string
-
-        path = environ.get("PATH_INFO", "")
-        if path:
-            attributes["url.path"] = path
-
-        url_full = get_request_url(environ, use_x_forwarded_for)
-        if query_string:
-            url_full += "?" + query_string
-        attributes["url.full"] = url_full
-
     return attributes
diff --git a/tests/integrations/wsgi/test_wsgi.py b/tests/integrations/wsgi/test_wsgi.py
index 12fed06ff8..4ec26dd405 100644
--- a/tests/integrations/wsgi/test_wsgi.py
+++ b/tests/integrations/wsgi/test_wsgi.py
@@ -11,7 +11,6 @@
     _ScopedResponse,
     get_request_url,
 )
-from tests.integrations.utils import DATA_COLLECTION_USER_INFO_CASES
 
 
 @pytest.fixture
@@ -45,7 +44,7 @@ def next(self):
 
 
 def test_basic(sentry_init, crashing_app, capture_events):
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     app = SentryWsgiMiddleware(crashing_app)
     client = Client(app)
     events = capture_events()
@@ -60,7 +59,6 @@ def test_basic(sentry_init, crashing_app, capture_events):
         "env": {"SERVER_NAME": "localhost", "SERVER_PORT": "80"},
         "headers": {"Host": "localhost"},
         "method": "GET",
-        "query_string": "",
         "url": "http://localhost/",
     }
 
@@ -70,7 +68,7 @@ def test_basic(sentry_init, crashing_app, capture_events):
 def test_script_name_is_respected(
     sentry_init, crashing_app, capture_events, script_name, path_info
 ):
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     app = SentryWsgiMiddleware(crashing_app)
     client = Client(app)
     events = capture_events()
@@ -87,7 +85,7 @@ def test_script_name_is_respected(
 @pytest.fixture(params=[0, None])
 def test_systemexit_zero_is_ignored(sentry_init, capture_events, request):
     zero_code = request.param
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     iterable = ExitingIterable(lambda: SystemExit(zero_code))
     app = SentryWsgiMiddleware(IterableApp(iterable))
     client = Client(app)
@@ -102,7 +100,7 @@ def test_systemexit_zero_is_ignored(sentry_init, capture_events, request):
 @pytest.fixture(params=["", "foo", 1, 2])
 def test_systemexit_nonzero_is_captured(sentry_init, capture_events, request):
     nonzero_code = request.param
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     iterable = ExitingIterable(lambda: SystemExit(nonzero_code))
     app = SentryWsgiMiddleware(IterableApp(iterable))
     client = Client(app)
@@ -121,7 +119,7 @@ def test_systemexit_nonzero_is_captured(sentry_init, capture_events, request):
 
 
 def test_keyboard_interrupt_is_captured(sentry_init, capture_events):
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     iterable = ExitingIterable(lambda: KeyboardInterrupt())
     app = SentryWsgiMiddleware(IterableApp(iterable))
     client = Client(app)
@@ -150,7 +148,7 @@ def dogpark(environ, start_response):
         raise ValueError("Fetch aborted. The ball was not returned.")
 
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sample_rate=1.0,
     )
     app = SentryWsgiMiddleware(dogpark)
@@ -184,20 +182,18 @@ def dogpark(environ, start_response):
     assert span_item["status"] == "error"
 
 
-@pytest.mark.parametrize("send_pii", [True, False])
 def test_transaction_no_error(
     sentry_init,
     capture_events,
     capture_items,
     DictionaryContaining,  # noqa:N803
-    send_pii,
 ):
     def dogpark(environ, start_response):
         start_response("200 OK", [])
         return ["Go get the ball! Good dog!"]
 
     sentry_init(
-        send_default_pii=send_pii,
+        data_collection={},
         traces_sample_rate=1.0,
     )
     app = SentryWsgiMiddleware(dogpark)
@@ -220,17 +216,12 @@ def dogpark(environ, start_response):
     assert span["attributes"]["http.response.status_code"] == 200
     assert span["status"] == "ok"
 
-    if send_pii:
-        assert (
-            span["attributes"]["url.full"]
-            == "http://localhost/dogs/are/great?toy=tennisball"
-        )
-        assert span["attributes"]["url.path"] == "/dogs/are/great"
-        assert span["attributes"]["http.query"] == "toy=tennisball"
-    else:
-        assert "url.path" not in span["attributes"]
-        assert "url.full" not in span["attributes"]
-        assert "http.query" not in span["attributes"]
+    assert (
+        span["attributes"]["url.full"]
+        == "http://localhost/dogs/are/great?toy=tennisball"
+    )
+    assert span["attributes"]["url.path"] == "/dogs/are/great"
+    assert span["attributes"]["http.query"] == "toy=tennisball"
 
 
 def test_has_trace_if_performance_enabled(
@@ -381,7 +372,7 @@ def app(environ, start_response):
 
     traces_sampler = mock.Mock(return_value=True)
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sampler=traces_sampler,
     )
     app = SentryWsgiMiddleware(app)
@@ -424,7 +415,7 @@ def app(environ, start_response):
 
     traces_sampler = mock.Mock(return_value=True)
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sampler=traces_sampler,
     )
     app = SentryWsgiMiddleware(app)
@@ -466,7 +457,7 @@ def sample_app(environ, start_response):
 
     traces_sampler = mock.Mock(return_value=True)
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sampler=traces_sampler,
     )
     app = SentryWsgiMiddleware(sample_app)
@@ -508,7 +499,7 @@ def dogpark(environ, start_response):
         return ["Go get the ball! Good dog!"]
 
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sample_rate=1.0,
     )
     app = SentryWsgiMiddleware(dogpark)
@@ -530,7 +521,7 @@ def dogpark(environ, start_response):
         return ["Go get the ball! Good dog!"]
 
     sentry_init(
-        send_default_pii=True,
+        data_collection={},
         traces_sample_rate=1.0,
     )
     app = SentryWsgiMiddleware(
@@ -687,21 +678,15 @@ def test_get_request_url_x_forwarded_proto(environ, use_x_forwarded_for, expecte
     assert get_request_url(environ, use_x_forwarded_for) == expected_url
 
 
-@pytest.mark.parametrize("send_default_pii", [True, False])
 def test_request_headers_data_collection_default_redacts_sensitive(
-    sentry_init, crashing_app, capture_events, send_default_pii
+    sentry_init, crashing_app, capture_events
 ):
     """
     When ``data_collection`` is configured (here as ``{}``, i.e. spec
     defaults), the WSGI event processor routes request headers through the
-    data-collection filtering path. Sensitive headers are redacted regardless
-    of ``send_default_pii`` -- the value of that legacy option must not change
-    the outcome.
+    data-collection filtering path. Sensitive headers are redacted.
     """
-    sentry_init(
-        send_default_pii=send_default_pii,
-        data_collection={},
-    )
+    sentry_init(data_collection={})
     app = SentryWsgiMiddleware(crashing_app)
     client = Client(app)
     events = capture_events()
@@ -722,22 +707,15 @@ def test_request_headers_data_collection_default_redacts_sensitive(
     assert headers["X-Custom-Header"] == "passthrough"
 
 
-def test_request_headers_legacy_no_pii_redacts_sensitive(
+def test_request_headers_data_collection_redacts_forwarded_headers(
     sentry_init, crashing_app, capture_events
 ):
-    """
-    With no ``data_collection`` configured, ``_filter_headers`` falls back to
-    the legacy ``send_default_pii`` behaviour. When PII is disabled, headers in
-    ``SENSITIVE_HEADERS`` are replaced with an ``AnnotatedValue`` (the default
-    ``use_annotated_value=True`` on the event-processor call site), which
-    serializes to an emptied value plus a ``_meta`` annotation. Non-sensitive
-    headers pass through untouched.
-
-    ``X-Forwarded-For`` is used because it is in ``SENSITIVE_HEADERS`` but is
-    not scrubbed by the default ``EventScrubber``, so the substitution we are
-    asserting on can only come from ``_filter_headers``.
-    """
-    sentry_init(send_default_pii=False)
+
+    sentry_init(
+        data_collection={
+            "http_headers": {"request": {"mode": "denylist", "terms": ["forwarded"]}}
+        }
+    )
     app = SentryWsgiMiddleware(crashing_app)
     client = Client(app)
     events = capture_events()
@@ -753,15 +731,9 @@ def test_request_headers_legacy_no_pii_redacts_sensitive(
 
     (event,) = events
 
-    assert event["request"]["headers"]["X-Forwarded-For"] == ""
+    assert event["request"]["headers"]["X-Forwarded-For"] == "[Filtered]"
     assert event["request"]["headers"]["X-Custom-Header"] == "passthrough"
 
-    # The emptied value is accompanied by a `_meta` annotation marking it as
-    # removed, confirming the substitution came from the AnnotatedValue path.
-    assert event["_meta"]["request"]["headers"]["X-Forwarded-For"] == {
-        "": {"rem": [["!config", "x"]]}
-    }
-
 
 def test_request_headers_data_collection_off_collects_no_headers(
     sentry_init, crashing_app, capture_events
@@ -900,15 +872,10 @@ def test_request_headers_data_collection_cookie_always_redacted(
     assert headers["X-Custom-Header"] == "passthrough"
 
 
-def test_request_headers_legacy_pii_passes_headers_through(
+def test_request_headers_data_collection_default_passes_non_sensitive_headers(
     sentry_init, crashing_app, capture_events
 ):
-    """
-    With no ``data_collection`` configured and ``send_default_pii`` enabled,
-    the legacy path returns all headers unchanged -- including those in
-    ``SENSITIVE_HEADERS``.
-    """
-    sentry_init(send_default_pii=True)
+    sentry_init(data_collection={})
     app = SentryWsgiMiddleware(crashing_app)
     client = Client(app)
     events = capture_events()
@@ -932,21 +899,10 @@ def test_request_headers_legacy_pii_passes_headers_through(
 @pytest.mark.parametrize(
     "init_kwargs, expected_query_string",
     [
-        # No data_collection: the legacy path always sets the query string
-        # unchanged, regardless of send_default_pii.
-        pytest.param(
-            {"send_default_pii": True},
-            "toy=tennisball&color=red&auth=secret",
-            id="send_default_pii_true",
-        ),
-        pytest.param(
-            {"send_default_pii": False},
-            "toy=tennisball&color=red&auth=secret",
-            id="send_default_pii_false",
-        ),
+        # data_collection omitted: use default denylist.
         pytest.param(
             {},
-            "toy=tennisball&color=red&auth=secret",
+            "toy=tennisball&color=red&auth=%5BFiltered%5D",
             id="defaults",
         ),
         # data_collection configured: query string is routed through filtering.
@@ -1005,21 +961,10 @@ def test_query_string_data_collection(
 @pytest.mark.parametrize(
     "init_kwargs, expected_query",
     [
-        # No data_collection: the ``http.query`` attribute follows the legacy
-        # send_default_pii gate.
-        pytest.param(
-            {"send_default_pii": True},
-            "toy=tennisball&color=red&auth=secret",
-            id="send_default_pii_true",
-        ),
-        pytest.param(
-            {"send_default_pii": False},
-            None,
-            id="send_default_pii_false",
-        ),
+        # data_collection omitted: use default denylist.
         pytest.param(
             {},
-            None,
+            "toy=tennisball&color=red&auth=%5BFiltered%5D",
             id="defaults",
         ),
         # data_collection configured: attribute is routed through filtering.
@@ -1083,38 +1028,21 @@ def dogpark(environ, start_response):
         assert span["attributes"]["http.query"] == expected_query
 
 
-@pytest.mark.parametrize("send_default_pii", [True, False])
-def test_user_ip_address_on_all_spans(sentry_init, capture_items, send_default_pii):
-    def dogpark(environ, start_response):
-        with sentry_sdk.start_span(name="child-span"):
-            pass
-        start_response("200 OK", [])
-        return ["Go get the ball! Good dog!"]
-
-    sentry_init(
-        send_default_pii=send_default_pii,
-        traces_sample_rate=1.0,
-    )
-    app = SentryWsgiMiddleware(dogpark)
-    client = Client(app)
-
-    items = capture_items("span")
-
-    client.get("/dogs/are/great/", environ_base={"REMOTE_ADDR": "127.0.0.1"})
-
-    sentry_sdk.flush()
-
-    child_span, server_span = [item.payload for item in items]
-
-    if send_default_pii:
-        assert server_span["attributes"]["user.ip_address"] == "127.0.0.1"
-        assert child_span["attributes"]["user.ip_address"] == "127.0.0.1"
-    else:
-        assert "user.ip_address" not in server_span["attributes"]
-        assert "user.ip_address" not in child_span["attributes"]
-
-
-@pytest.mark.parametrize("init_kwargs, expect_ip", DATA_COLLECTION_USER_INFO_CASES)
+@pytest.mark.parametrize(
+    "init_kwargs, expect_ip",
+    [
+        pytest.param(
+            {"data_collection": {"user_info": True}},
+            True,
+            id="data_collection_user_info_true",
+        ),
+        pytest.param(
+            {"data_collection": {"user_info": False}},
+            False,
+            id="data_collection_user_info_false",
+        ),
+    ],
+)
 def test_user_info_span_attributes_data_collection(
     sentry_init, capture_items, init_kwargs, expect_ip
 ):
@@ -1151,7 +1079,21 @@ def dogpark(environ, start_response):
         assert "client.address" not in server_span["attributes"]
 
 
-@pytest.mark.parametrize("init_kwargs, expect_ip", DATA_COLLECTION_USER_INFO_CASES)
+@pytest.mark.parametrize(
+    "init_kwargs, expect_ip",
+    [
+        pytest.param(
+            {"data_collection": {"user_info": True}},
+            True,
+            id="data_collection_user_info_true",
+        ),
+        pytest.param(
+            {"data_collection": {"user_info": False}},
+            False,
+            id="data_collection_user_info_false",
+        ),
+    ],
+)
 def test_user_info_error_event_data_collection(
     sentry_init, crashing_app, capture_events, init_kwargs, expect_ip
 ):