diff --git a/sentry_sdk/integrations/flask.py b/sentry_sdk/integrations/flask.py
index 1902091fbf..4baa65b183 100644
--- a/sentry_sdk/integrations/flask.py
+++ b/sentry_sdk/integrations/flask.py
@@ -13,6 +13,7 @@
     capture_internal_exceptions,
     ensure_integration_enabled,
     event_from_exception,
+    has_data_collection_enabled,
     package_version,
 )
 
@@ -142,7 +143,8 @@ def _set_transaction_name_and_source(
 
 
 def _request_started(app: "Flask", **kwargs: "Any") -> None:
-    integration = sentry_sdk.get_client().get_integration(FlaskIntegration)
+    client = sentry_sdk.get_client()
+    integration = client.get_integration(FlaskIntegration)
     if integration is None:
         return
 
@@ -156,7 +158,13 @@ def _request_started(app: "Flask", **kwargs: "Any") -> None:
 
     scope = sentry_sdk.get_isolation_scope()
 
-    if should_send_default_pii():
+    if has_data_collection_enabled(client.options):
+        if client.options["data_collection"]["user_info"]:
+            with capture_internal_exceptions():
+                user_properties = _get_flask_user_properties()
+                if user_properties:
+                    scope.set_user(user_properties)
+    elif should_send_default_pii():
         with capture_internal_exceptions():
             user_properties = _get_flask_user_properties()
             if user_properties:
@@ -208,7 +216,12 @@ def inner(event: "Event", hint: "dict[str, Any]") -> "Event":
         with capture_internal_exceptions():
             FlaskRequestExtractor(request).extract_into_event(event)
 
-        if should_send_default_pii():
+        client_options = sentry_sdk.get_client().options
+        if has_data_collection_enabled(client_options):
+            if client_options["data_collection"]["user_info"]:
+                with capture_internal_exceptions():
+                    _add_user_to_event(event)
+        elif should_send_default_pii():
             with capture_internal_exceptions():
                 _add_user_to_event(event)
 
diff --git a/sentry_sdk/scope.py b/sentry_sdk/scope.py
index c0b53bfaa4..4fa8687569 100644
--- a/sentry_sdk/scope.py
+++ b/sentry_sdk/scope.py
@@ -57,6 +57,7 @@
     event_from_exception,
     exc_info_from_error,
     format_attribute,
+    has_data_collection_enabled,
     has_logs_enabled,
     has_metrics_enabled,
     logger,
@@ -1775,7 +1776,11 @@ def _apply_user_attributes_to_telemetry(
         else:
             attributes = telemetry._attributes
 
-        if not should_send_default_pii() or self._user is None:
+        client_options = sentry_sdk.get_client().options
+        if has_data_collection_enabled(client_options):
+            if not client_options["data_collection"]["user_info"] or self._user is None:
+                return
+        elif not should_send_default_pii() or self._user is None:
             return
 
         for attribute_name, user_attribute in (
diff --git a/tests/integrations/flask/test_flask.py b/tests/integrations/flask/test_flask.py
index a0728bf226..c091e23f61 100644
--- a/tests/integrations/flask/test_flask.py
+++ b/tests/integrations/flask/test_flask.py
@@ -1521,3 +1521,105 @@ def crash():
     (event,) = events
 
     assert "ip_address" not in event.get("user", {})
+
+
+@pytest.mark.parametrize("init_kwargs, expect_user", USER_INFO_INIT_KWARGS)
+def test_flask_login_user_identity_error_event_data_collection(
+    sentry_init, app, capture_events, init_kwargs, expect_user
+):
+    sentry_init(integrations=[flask_sentry.FlaskIntegration()], **init_kwargs)
+
+    class User:
+        is_authenticated = is_active = True
+        is_anonymous = False
+        email = "user@example.com"
+        username = "testuser"
+
+        def get_id(self):
+            return "42"
+
+    @login_manager.user_loader
+    def load_user(user_id):
+        return User()
+
+    @app.route("/login")
+    def login():
+        login_user(User())
+        return "ok"
+
+    @app.route("/crash")
+    def crash():
+        1 / 0
+
+    events = capture_events()
+
+    client = app.test_client()
+    assert client.get("/login").status_code == 200
+    with pytest.raises(ZeroDivisionError):
+        client.get("/crash")
+
+    (event,) = events
+
+    if expect_user:
+        assert event["user"]["id"] == "42"
+        assert event["user"]["email"] == "user@example.com"
+        assert event["user"]["username"] == "testuser"
+    else:
+        user = event.get("user", {})
+        assert "id" not in user
+        assert "email" not in user
+        assert "username" not in user
+
+
+@pytest.mark.parametrize("init_kwargs, expect_user", USER_INFO_INIT_KWARGS)
+def test_flask_login_user_identity_span_attributes_data_collection(
+    sentry_init, app, capture_items, init_kwargs, expect_user
+):
+    init_kwargs = dict(init_kwargs)
+    experiments = init_kwargs.pop("_experiments", {})
+
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        trace_lifecycle="stream",
+        _experiments=experiments,
+        **init_kwargs,
+    )
+
+    class User:
+        is_authenticated = is_active = True
+        is_anonymous = False
+        email = "user@example.com"
+        username = "testuser"
+
+        def get_id(self):
+            return "42"
+
+    @login_manager.user_loader
+    def load_user(user_id):
+        return User()
+
+    @app.route("/login")
+    def login():
+        login_user(User())
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    assert client.get("/login").status_code == 200
+    assert client.get("/message").status_code == 200
+
+    sentry_sdk.flush()
+
+    spans = [item.payload for item in items if item.type == "span"]
+    segment = next(s for s in spans if s["name"] == "hi")
+
+    if expect_user:
+        assert segment["attributes"]["user.id"] == "42"
+        assert segment["attributes"]["user.email"] == "user@example.com"
+        assert segment["attributes"]["user.name"] == "testuser"
+    else:
+        assert "user.id" not in segment.get("attributes", {})
+        assert "user.email" not in segment.get("attributes", {})
+        assert "user.name" not in segment.get("attributes", {})