@@ -1065,6 +1065,19 @@ def has_operations_mixin(self) -> bool:
             > 0
         )
 
+    @cached_property
+    def has_resumable_upload_methods(self) -> bool:
+        """Return whether any service in the API has resumable upload methods.
+
+        Used in package-level templates (such as setup.py.j2) where the full API
+        is in scope rather than an individual service, to determine whether the
+        generated package requires resumable upload dependencies.
+        """
+        return any(
+            service.has_resumable_upload_methods
+            for service in self.services.values()
+        )
+
     @cached_property
     def _has_iam_overrides(self) -> bool:
         if not self.has_iam_mixin:
@@ -1651,6 +1664,7 @@ def _get_methods(
                 output=self.api_messages[meth_pb.output_type.lstrip(".")],
                 retry=retry,
                 timeout=timeout,
+                resumable_upload_prefix=self.opts.resumable_upload_prefix,
             )
 
         # Done; return the answer.
 
@@ -29,7 +29,9 @@ def __str__(self) -> str:
         # Determine if we need to suppress type checking for this import.
         # We do this for protobuf generated files (_pb2) and api_core
         # internals where type information might be missing or incomplete.
-        needs_type_ignore = self.module.endswith("_pb2") or "api_core" in self.package
+        needs_type_ignore = self.module.endswith("_pb2") or (
+            "api_core" in self.package and self.module != "resumable_transfer"
+        )
         if needs_type_ignore:
             # Use 'import absolute.path as module' syntax to prevent Ruff/isort
             # from combining this with other imports. This ensures the
 
@@ -1499,6 +1499,7 @@ class Method:
     meta: metadata.Metadata = dataclasses.field(
         default_factory=metadata.Metadata,
     )
+    resumable_upload_prefix: str = "resumable/upload"
 
     def __getattr__(self, name):
         return getattr(self.method_pb, name)
@@ -1728,6 +1729,13 @@ def http_opt(self) -> Optional[Dict[str, str]]:
         # TODO(yon-mg): enums for http verbs?
         return answer
 
+    @property
+    def is_resumable_upload(self) -> bool:
+        """Return True if this method is a resumable upload method."""
+        # Resumable upload method names are temporarily hardcoded here until
+        # the resumable upload proto annotation exists and is published.
+        return self.name in ("UploadMedia", "CreateYouTubeVideoUpload")
+
     @property
     def path_params(self) -> Sequence[str]:
         """Return the path parameters found in the http annotation path template"""
@@ -2208,6 +2216,16 @@ def has_pagers(self) -> bool:
         """Return whether the service has paged methods."""
         return any(m.paged_result_field for m in self.methods.values())
 
+    @property
+    def has_resumable_upload_methods(self) -> bool:
+        """Return whether this specific service has resumable upload methods.
+
+        Used during per-service file generation and in service-level templates
+        to conditionally emit resumable upload imports, transports, and client
+        methods only for services that define resumable upload RPCs.
+        """
+        return any(m.is_resumable_upload for m in self.methods.values())
+
     @property
     def host(self) -> str:
         """Return the hostname for this service, if specified.
 
@@ -52,6 +52,7 @@ class Options:
     proto_plus_deps: Tuple[str, ...] = dataclasses.field(default=("",))
     gapic_version: str = "0.0.0"
     resource_name_aliases: Dict[str, str] = dataclasses.field(default_factory=dict)
+    resumable_upload_prefix: str = "resumable/upload"
 
     # Class constants
     PYTHON_GAPIC_PREFIX: str = "python-gapic-"
@@ -78,6 +79,8 @@ class Options:
             # resource path to a custom TitleCase alias.
             # Format: resource.path/Name:AliasName
             "resource-name-alias",
+            # Prefix for resumable upload requests
+            "resumable-upload-prefix",
         )
     )
 
@@ -222,6 +225,8 @@ def tweak_path(p):
                     "Expected format is 'resource.path/Name:AliasName'."
                 )
 
+        resumable_upload_prefix = opts.pop("resumable-upload-prefix", ["resumable/upload"]).pop()
+
         answer = Options(
             name=opts.pop("name", [""]).pop(),
             namespace=tuple(opts.pop("namespace", [])),
@@ -245,6 +250,7 @@ def tweak_path(p):
             proto_plus_deps=proto_plus_deps,
             gapic_version=opts.pop("gapic-version", ["0.0.0"]).pop(),
             resource_name_aliases=resource_name_aliases,
+            resumable_upload_prefix=resumable_upload_prefix,
         )
 
         # Note: if we ever need to recursively check directories for sample
 
@@ -171,6 +171,7 @@ def test_echo_api_version_header(transport_name):
                 type(client.transport.echo),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.EchoResponse()
                 client.echo()
 
         # Establish that the api version header was sent.
@@ -196,6 +197,7 @@ def test_echo_error_details_api_version_header(transport_name):
                 type(client.transport.echo_error_details),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.EchoErrorDetailsResponse()
                 client.echo_error_details()
 
         # Establish that the api version header was sent.
@@ -221,6 +223,7 @@ def test_fail_echo_with_details_api_version_header(transport_name):
                 type(client.transport.fail_echo_with_details),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.FailEchoWithDetailsResponse()
                 client.fail_echo_with_details()
 
         # Establish that the api version header was sent.
@@ -246,6 +249,7 @@ def test_expand_api_version_header(transport_name):
                 type(client.transport.expand),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.EchoResponse()
                 client.expand()
 
         # Establish that the api version header was sent.
@@ -271,6 +275,7 @@ def test_collect_api_version_header(transport_name):
                 type(client.transport.collect),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.EchoResponse()
                 client.collect()
 
         # Establish that the api version header was sent.
@@ -296,6 +301,7 @@ def test_chat_api_version_header(transport_name):
                 type(client.transport.chat),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.EchoResponse()
                 client.chat()
 
         # Establish that the api version header was sent.
@@ -321,6 +327,7 @@ def test_paged_expand_api_version_header(transport_name):
                 type(client.transport.paged_expand),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.PagedExpandResponse()
                 client.paged_expand()
 
         # Establish that the api version header was sent.
@@ -346,6 +353,7 @@ def test_paged_expand_legacy_api_version_header(transport_name):
                 type(client.transport.paged_expand_legacy),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.PagedExpandResponse()
                 client.paged_expand_legacy()
 
         # Establish that the api version header was sent.
@@ -371,6 +379,7 @@ def test_paged_expand_legacy_mapped_api_version_header(transport_name):
                 type(client.transport.paged_expand_legacy_mapped),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.PagedExpandLegacyMappedResponse()
                 client.paged_expand_legacy_mapped()
 
         # Establish that the api version header was sent.
@@ -396,6 +405,7 @@ def test_wait_api_version_header(transport_name):
                 type(client.transport.wait),
                 '__call__'
             ) as call:
+                call.return_value = operations_pb2.Operation()
                 client.wait()
 
         # Establish that the api version header was sent.
@@ -421,6 +431,7 @@ def test_block_api_version_header(transport_name):
                 type(client.transport.block),
                 '__call__'
             ) as call:
+                call.return_value = gs_echo.BlockResponse()
                 client.block()
 
         # Establish that the api version header was sent.
 
@@ -42,6 +42,7 @@ class DummyMethod:
     client_output: bool = False
     client_output_async: bool = False
     is_internal: bool = False
+    is_resumable_upload: bool = False
 
     @property
     def client_method_name(self):
 
@@ -280,3 +280,14 @@ def test_options_resource_name_aliases():
         # 3. MissingAlias: 
         # (The empty ' ' string safely 'continues' without warning, as intended)
         assert warn.call_count == 3
+
+
+def test_options_resumable_upload_prefix():
+    # Verify that resumable_upload_prefix defaults to "resumable/upload" when not specified.
+    opts_default = Options.build("")
+    assert opts_default.resumable_upload_prefix == "resumable/upload"
+
+    # Verify that a custom resumable-upload-prefix option is parsed and stored.
+    opts_custom = Options.build("resumable-upload-prefix=custom/upload/prefix")
+    assert opts_custom.resumable_upload_prefix == "custom/upload/prefix"
+
@@ -4745,3 +4745,31 @@ def test_api_build_selective_multiple_protos_kept():
     assert len(api_schema.protos) == 2
     assert "proto1.proto" in api_schema.protos
     assert "proto2.proto" in api_schema.protos
+    assert not api_schema.has_resumable_upload_methods
+
+
+def test_api_has_resumable_upload_methods():
+    # Verify that an API containing a recognized resumable upload method
+    # reports has_resumable_upload_methods as True.
+    # Note: In MethodDescriptorProto, output_type is the protobuf RPC response
+    # message defined in the .proto file (e.g. UploadResponse).
+    fd = make_file_pb2(
+        name="upload.proto",
+        package="google.example.v1",
+        services=(
+            descriptor_pb2.ServiceDescriptorProto(
+                name="UploadService",
+                method=(
+                    descriptor_pb2.MethodDescriptorProto(
+                        name="UploadMedia",
+                        input_type="google.example.v1.UploadRequest",
+                        output_type="google.example.v1.UploadResponse",
+                    ),
+                ),
+            ),
+        ),
+        messages=(make_message_pb2("UploadRequest"), make_message_pb2("UploadResponse")),
+    )
+    api_schema = api.API.build([fd], package="google.example.v1")
+    assert api_schema.has_resumable_upload_methods
+
@@ -40,6 +40,12 @@ def test_str_untyped_api_core():
     assert str(i) == "import foo.api_core.baz as bacon  # type: ignore"
 
 
+def test_str_typed_resumable_transfer():
+    # Verify that google.api_core.resumable_transfer is imported without "# type: ignore".
+    i = imp.Import(package=("google", "api_core"), module="resumable_transfer")
+    assert str(i) == "from google.api_core import resumable_transfer"
+
+
 def test_str_eq():
     i1 = imp.Import(package=("foo", "bar"), module="baz")
     i2 = imp.Import(package=("foo", "bar"), module="baz")
 
@@ -1118,3 +1118,21 @@ def test__validate_paged_field_size_type(field_type, pb_type, expected):
 
     actual = method._validate_paged_field_size_type(page_field_size=page_size)
     assert actual == expected
+
+
+def test_method_is_resumable_upload():
+    # UploadMedia (and CreateYouTubeVideoUpload) are temporarily hardcoded as
+    # resumable upload method names until resumable upload proto annotations
+    # are adopted in the future.
+    # Verify that UploadMedia is identified as a resumable upload method.
+    method_upload = make_method("UploadMedia")
+    assert method_upload.is_resumable_upload
+
+    # Verify that CreateYouTubeVideoUpload is also recognized as a resumable upload method.
+    method_youtube_upload = make_method("CreateYouTubeVideoUpload")
+    assert method_youtube_upload.is_resumable_upload
+
+    # Verify that a non-resumable method name is not marked as a resumable upload.
+    method_other = make_method("OtherMethod")
+    assert not method_other.is_resumable_upload
+