fix: lazy-import boto3 in serverless io_manager so v2 user pods don't need it (#25605)

## Summary

The serverless v2 (k8s) IO manager (`ServerlessPresignedURLIOManager`)
does not use boto3 — it talks to S3 indirectly via presigned URLs handed
out by the host cloud API. But `dagster_cloud.serverless.io_manager`
still imports boto3 at module load time for the legacy ECS-based
`PickledObjectServerlessIOManager`. This blows up on v2 user pods built
from a plain `dagster-cloud` dependency (no `[serverless]` extra), which
is what the `dg`-scaffolded projects ship.

Move the boto3 import inside
`PickledObjectServerlessIOManager._refresh_boto_session`. The ECS path
still loads boto3 the first time it instantiates the legacy class; the
v2 path now loads `dagster_cloud.serverless` cleanly without boto3
installed.

Internal-RevId: 697f3c50c156f6ea2977cc0087c1c7e5d515b008

Author: sndagster

python_modules/dagster-cloud/dagster_cloud/serverless/io_manager.py

@@ -2,9 +2,8 @@
 import io
 import os
 import pickle
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-import boto3
 import dagster._check as check
 import requests
 from dagster import InputContext, OutputContext, UPathIOManager, io_manager
@@ -21,6 +20,9 @@
 from upath import UPath
 from urllib3.util.retry import Retry
 
+if TYPE_CHECKING:
+    import boto3
+
 ECS_AGENT_IP = "169.254.170.2"
 
 
@@ -36,7 +38,11 @@ def __init__(
         base_path = UPath(s3_prefix) if s3_prefix else None
         super().__init__(base_path=base_path)
 
-    def _refresh_boto_session(self) -> tuple[boto3.Session, datetime.datetime]:
+    def _refresh_boto_session(self) -> tuple["boto3.Session", datetime.datetime]:
+        # boto3 is lazy-imported so this module can load in environments where
+        # only the presigned-URL IO manager path is used and boto3 is not installed.
+        import boto3
+
         # We have to do this whacky way to get credentials to ensure that we get iam role
         # we assigned to the task. If we used the default boto behavior, it could get overriden
         # when users set AWS env vars.

python_modules/dagster-cloud/dagster_cloud_tests/serverless_tests/test_io_manager.py

@@ -1,5 +1,7 @@
 # ruff: noqa: SLF001
+import importlib
 import pickle
+import sys
 from unittest import mock
 
 import pytest
@@ -301,3 +303,44 @@ def test_serverless_io_manager_usable_in_job():
         resources={"io_manager": serverless_io_manager},
     )
     assert defs
+
+
+def test_module_imports_without_boto3_installed():
+    # Simulates a v2 user pod whose image only depends on dagster-cloud (no
+    # [serverless] extra) and therefore has no boto3 installed. The module must
+    # still load cleanly so the presigned-URL IO manager path can be selected
+    # at runtime via SERVERLESS_IO_MANAGER_USE_PRESIGNED_URL.
+    cached = {
+        name: sys.modules.pop(name)
+        for name in (
+            "boto3",
+            "boto3.s3",
+            "boto3.s3.transfer",
+            "boto3.session",
+            "botocore",
+            "dagster_cloud.serverless",
+            "dagster_cloud.serverless.io_manager",
+        )
+        if name in sys.modules
+    }
+    try:
+        # `None` in sys.modules is the standard poison-pill that forces
+        # `import boto3` to raise ImportError.
+        with mock.patch.dict(sys.modules, {"boto3": None}):
+            reloaded = importlib.import_module("dagster_cloud.serverless.io_manager")
+            assert reloaded.ServerlessPresignedURLIOManager is not None
+            # Instantiating the presigned-URL manager must not touch boto3.
+            manager = reloaded.ServerlessPresignedURLIOManager(
+                api_url=API_URL, api_token=API_TOKEN, timeout=30
+            )
+            assert manager is not None
+    finally:
+        # Drop anything we re-imported under the poison-pill and restore the
+        # originals so other tests see a clean module state.
+        for name in (
+            "dagster_cloud.serverless",
+            "dagster_cloud.serverless.io_manager",
+        ):
+            sys.modules.pop(name, None)
+        sys.modules.update(cached)
+        importlib.import_module("dagster_cloud.serverless.io_manager")