From 1d6684ade62ae87eb069c4ea2ebd83a626856325 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 29 Sep 2026 19:11:40 +0000
Subject: [PATCH 1/3] feat(gapic-generator): generate code samples for
 resumable upload RPCs

---
 .../gapic/samplegen/samplegen.py              |  17 +-
 .../templates/examples/feature_fragments.j2   | 140 ++++++++++++-
 .../gapic/templates/examples/sample.py.j2     |   6 +-
 .../resumable_upload_service/async_client.py  | 103 +++++++++-
 .../resumable_upload_service/client.py        | 103 +++++++++-
 ...mable_upload_service_upload_media_async.py | 109 ++++++++++-
 ...umable_upload_service_upload_media_sync.py | 107 +++++++++-
 ...ppet_metadata_google.showcase.v1beta1.json |  36 ++--
 .../tests/unit/samplegen/test_integration.py  | 184 ++++++++++++++++++
 9 files changed, 766 insertions(+), 39 deletions(-)

diff --git a/packages/gapic-generator/gapic/samplegen/samplegen.py b/packages/gapic-generator/gapic/samplegen/samplegen.py
index 6f103225ea04..cca1bf0b0a87 100644
--- a/packages/gapic-generator/gapic/samplegen/samplegen.py
+++ b/packages/gapic-generator/gapic/samplegen/samplegen.py
@@ -1183,14 +1183,20 @@ def _get_sample_imports(sample: Dict, rpc: wrappers.Method) -> List[str]:
     module_name = sample["module_name"]
     module_import = f"from {module_namespace} import {module_name}"
 
+    imports = [module_import]
     address = rpc.input.meta.address
     # This checks if the request message is part of the service proto package.
     # If not, we should try to include a separate import statement.
-    if address.proto_package.startswith(address.api_naming.proto_package):
-        return [module_import]
-    else:
-        request_import = str(address.python_import)
-        return sorted([module_import, request_import])
+    if not address.proto_package.startswith(address.api_naming.proto_package):
+        imports.append(str(address.python_import))
+
+    if rpc.is_resumable_upload:
+        imports.append(
+            "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+        )
+        imports.append("import io")
+
+    return sorted(imports)
 
 
 def generate_sample(
@@ -1222,6 +1228,7 @@ def generate_sample(
 
     calling_form = types.CallingForm.method_default(rpc)
     sample["is_internal"] = rpc.is_internal
+    sample["is_resumable_upload"] = rpc.is_resumable_upload
 
     v = Validator(rpc, api_schema)
     # Tweak some small aspects of the sample to set defaults for optional
diff --git a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2 b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
index 40786ef5946b..13d747e7a673 100644
--- a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
+++ b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
@@ -23,7 +23,11 @@
 # It may require modifications to work in your environment.
 
 # To install the latest published package dependency, execute the following:
+{% if sample.get("is_resumable_upload") and sample.transport in ("grpc-async", "rest-async") %}
+#   python3 -m pip install {{ sample.package_name }}[async_rest]
+{% else %}
 #   python3 -m pip install {{ sample.package_name }}
+{% endif %}
 {% endmacro %}
 
 {% macro print_string_formatting(string_list) %}
@@ -226,10 +230,10 @@ request=request
 {% endmacro %}
 
 
-{% macro render_method_call(sample, calling_form, calling_form_enum, transport) %}
+{% macro render_method_call(sample, calling_form, calling_form_enum, transport, include_config=False) %}
 {# Note: this doesn't deal with enums or unions #}
 {# LROs return operation objects and paged requests return pager objects #}
-{% if transport == "grpc-async" and calling_form not in
+{% if transport == "grpc-async" and not sample.get("is_resumable_upload") and calling_form not in
 [calling_form_enum.RequestPaged, calling_form_enum.RequestPagedAll] %}
 await{{ " "}}
 {%- endif -%}
@@ -238,14 +242,142 @@ await{{ " "}}
 client.{{ render_method_name(sample)|trim }}(requests=request_generator())
 {% else %}{# TODO: deal with flattening #}
 {# TODO: set up client streaming once some questions are answered #}
-client.{{ render_method_name(sample)|trim }}({{ render_request_params_unary(sample.request)|trim }})
+client.{{ render_method_name(sample)|trim }}({{ render_request_params_unary(sample.request)|trim }}{% if include_config %}, config=config{% endif %})
 {% endif %}
 {% endmacro %}
 
 
+{% macro render_resumable_upload_samples(sample, calling_form, calling_form_enum) %}
+{% set is_async = sample.transport in ("grpc-async", "rest-async") %}
+{% set async_prefix = "async " if is_async else "" %}
+{% set await_prefix = "await " if is_async else "" %}
+{% set fn_name = sample.rpc|snake_case|trim %}
+{% set fn_params = print_input_params(sample.request)|trim %}
+{% set method_call_with_config = render_method_call(sample, calling_form, calling_form_enum, sample.transport, include_config=True)|trim %}
+{% set method_call_no_config = render_method_call(sample, calling_form, calling_form_enum, sample.transport, include_config=False)|trim %}
+{{ async_prefix }}def sample_{{ fn_name }}({{ fn_params }}):
+    # Option 1: Upload a stream directly from start to completion.
+    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
+    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = {{ method_call_with_config }}
+
+    # Upload the entire stream directly and {% if is_async %}await{% else %}return{% endif %} the final response
+    stream = io.BytesIO(b"Example upload data")
+    {{ "response = " if sample.response else "" }}{{ await_prefix }}upload_session.upload(stream)
+    {% if sample.response %}
+
+    # Handle the response
+    {% for statement in sample.response %}
+    {{ dispatch_statement(statement)|trim }}
+    {% endfor %}
+    {% endif %}
+
+
+{{ async_prefix }}def sample_{{ fn_name }}_with_progress({{ fn_params }}):
+    # Option 2: Upload a stream while receiving progress updates per chunk.
+    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
+    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = {{ method_call_with_config }}
+
+    # Iterate over the upload to receive progress updates as each chunk is transmitted
+    stream = io.BytesIO(b"Example upload data")
+    progress: UploadProgress
+    {% if is_async %}
+    async for progress in upload_session.upload(stream):
+    {% else %}
+    for progress in upload_session.iter_upload(stream):
+    {% endif %}
+        print(
+            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+
+    # After iteration completes, the final response is available on the session
+    {{ "response = " if sample.response else "" }}upload_session.response
+    {% if sample.response %}
+
+    # Handle the response
+    {% for statement in sample.response %}
+    {{ dispatch_statement(statement)|trim }}
+    {% endfor %}
+    {% endif %}
+
+
+{{ async_prefix }}def sample_{{ fn_name }}_resume({{ fn_params }}):
+    # Option 3: Resume an interrupted upload using a previously saved session URL.
+    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
+    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
+    # Create an upload session for the request
+    upload_session = {{ method_call_no_config }}
+
+    # Resume the interrupted upload from the saved session URL and chunk size
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    {{ "response = " if sample.response else "" }}{{ await_prefix }}upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+    {% if sample.response %}
+
+    # Handle the response
+    {% for statement in sample.response %}
+    {{ dispatch_statement(statement)|trim }}
+    {% endfor %}
+    {% endif %}
+
+
+{{ async_prefix }}def sample_{{ fn_name }}_resume_with_progress({{ fn_params }}):
+    # Option 4: Resume an interrupted upload while receiving progress updates.
+    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
+    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
+    # Create an upload session for the request
+    upload_session = {{ method_call_no_config }}
+
+    # Resume the interrupted upload while iterating over progress updates
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    progress: UploadProgress
+    {% if is_async %}
+    async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
+    {% else %}
+    for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
+    {% endif %}
+        print(
+            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+        print(f"Chunk size: {progress.chunk_size}")
+
+    # After iteration completes, the final response is available on the session
+    {{ "response = " if sample.response else "" }}upload_session.response
+    {% if sample.response %}
+
+    # Handle the response
+    {% for statement in sample.response %}
+    {{ dispatch_statement(statement)|trim }}
+    {% endfor %}
+    {% endif %}
+{% endmacro %}
+
+
 {# Setting up the method invocation is the responsibility of the caller: #}
 {# it's just easier to set up client side streaming and other things from outside this macro. #}
-{% macro render_calling_form(method_invocation_text, calling_form, calling_form_enum, transport, response_statements ) %}
+{% macro render_calling_form(method_invocation_text, calling_form, calling_form_enum, transport, response_statements, sample=None ) %}
 # Make the request
 {% if calling_form in [calling_form_enum.Request, calling_form_enum.RequestStreamingClient] %}
 {% if response_statements %}response = {% endif %}{{ method_invocation_text|trim }}
diff --git a/packages/gapic-generator/gapic/templates/examples/sample.py.j2 b/packages/gapic-generator/gapic/templates/examples/sample.py.j2
index 68713f812b85..7b08a54e47c9 100644
--- a/packages/gapic-generator/gapic/templates/examples/sample.py.j2
+++ b/packages/gapic-generator/gapic/templates/examples/sample.py.j2
@@ -32,12 +32,16 @@
 
 
 {# also need calling form #}
+{% if sample.get("is_resumable_upload") %}
+{{ frags.render_resumable_upload_samples(sample, calling_form, calling_form_enum) }}
+{% else %}
 {% if sample.transport == "grpc-async" %}async {% endif %}def sample_{{ sample.rpc|snake_case|trim }}({{ frags.print_input_params(sample.request)|trim }}):
     {{ frags.render_client_setup(sample.module_name, sample.client_name)|indent }}
     {{ frags.render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
     {% with method_call = frags.render_method_call(sample, calling_form, calling_form_enum, sample.transport) %}
-    {{ frags.render_calling_form(method_call, calling_form, calling_form_enum, sample.transport, sample.response)|indent -}}
+    {{ frags.render_calling_form(method_call, calling_form, calling_form_enum, sample.transport, sample.response, sample)|indent -}}
     {% endwith %}
+{% endif %}
 
 # [END {{ sample.id }}]
 {# TODO: Enable main block (or decide to remove main block from python sample) #}
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
index 63b639314785..b3d88f76894a 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
@@ -280,8 +280,11 @@ async def upload_media(self,
             #   client as shown in:
             #   https://googleapis.dev/python/google-api-core/latest/client_options.html
             from google import showcase_v1beta1
+            from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+            import io
 
             async def sample_upload_media():
+                # Option 1: Upload a stream directly from start to completion.
                 # Create a client
                 client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
 
@@ -289,8 +292,104 @@ async def sample_upload_media():
                 request = showcase_v1beta1.UploadMediaRequest(
                 )
 
-                # Make the request
-                response = await client.upload_media(request=request)
+                # Configure optional transfer settings such as chunk size and stall detection
+                config = ResumableUploadConfig(
+                    chunk_size=8 * 1024 * 1024,  # 8 MB
+                    stall_minimum_rate=64 * 1024,
+                    stall_timeout=120,
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request, config=config)
+
+                # Upload the entire stream directly and await the final response
+                stream = io.BytesIO(b"Example upload data")
+                response = await upload_session.upload(stream)
+
+                # Handle the response
+                print(response)
+
+            async def sample_upload_media_with_progress():
+                # Option 2: Upload a stream while receiving progress updates per chunk.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Configure optional transfer settings such as chunk size and stall detection
+                config = ResumableUploadConfig(
+                    chunk_size=8 * 1024 * 1024,  # 8 MB
+                    stall_minimum_rate=64 * 1024,
+                    stall_timeout=120,
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request, config=config)
+
+                # Iterate over the upload to receive progress updates as each chunk is transmitted
+                stream = io.BytesIO(b"Example upload data")
+                progress: UploadProgress
+                async for progress in upload_session.upload(stream):
+                    print(
+                        f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+                    )
+                    print(f"Session URL: {progress.upload_url}")
+
+                # After iteration completes, the final response is available on the session
+                response = upload_session.response
+
+                # Handle the response
+                print(response)
+
+            async def sample_upload_media_resume():
+                # Option 3: Resume an interrupted upload using a previously saved session URL.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request)
+
+                # Resume the interrupted upload from the saved session URL and chunk size
+                stream = io.BytesIO(b"Example upload data")
+                upload_url = "https://..."
+                chunk_size = 8 * 1024 * 1024
+                response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+
+                # Handle the response
+                print(response)
+
+            async def sample_upload_media_resume_with_progress():
+                # Option 4: Resume an interrupted upload while receiving progress updates.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request)
+
+                # Resume the interrupted upload while iterating over progress updates
+                stream = io.BytesIO(b"Example upload data")
+                upload_url = "https://..."
+                chunk_size = 8 * 1024 * 1024
+                progress: UploadProgress
+                async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
+                    print(
+                        f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+                    )
+                    print(f"Session URL: {progress.upload_url}")
+                    print(f"Chunk size: {progress.chunk_size}")
+
+                # After iteration completes, the final response is available on the session
+                response = upload_session.response
 
                 # Handle the response
                 print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
index 4fedb635bc91..2626bd662a3e 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
@@ -514,8 +514,11 @@ def upload_media(self,
             #   client as shown in:
             #   https://googleapis.dev/python/google-api-core/latest/client_options.html
             from google import showcase_v1beta1
+            from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+            import io
 
             def sample_upload_media():
+                # Option 1: Upload a stream directly from start to completion.
                 # Create a client
                 client = showcase_v1beta1.ResumableUploadServiceClient()
 
@@ -523,8 +526,104 @@ def sample_upload_media():
                 request = showcase_v1beta1.UploadMediaRequest(
                 )
 
-                # Make the request
-                response = client.upload_media(request=request)
+                # Configure optional transfer settings such as chunk size and stall detection
+                config = ResumableUploadConfig(
+                    chunk_size=8 * 1024 * 1024,  # 8 MB
+                    stall_minimum_rate=64 * 1024,
+                    stall_timeout=120,
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request, config=config)
+
+                # Upload the entire stream directly and return the final response
+                stream = io.BytesIO(b"Example upload data")
+                response = upload_session.upload(stream)
+
+                # Handle the response
+                print(response)
+
+            def sample_upload_media_with_progress():
+                # Option 2: Upload a stream while receiving progress updates per chunk.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Configure optional transfer settings such as chunk size and stall detection
+                config = ResumableUploadConfig(
+                    chunk_size=8 * 1024 * 1024,  # 8 MB
+                    stall_minimum_rate=64 * 1024,
+                    stall_timeout=120,
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request, config=config)
+
+                # Iterate over the upload to receive progress updates as each chunk is transmitted
+                stream = io.BytesIO(b"Example upload data")
+                progress: UploadProgress
+                for progress in upload_session.iter_upload(stream):
+                    print(
+                        f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+                    )
+                    print(f"Session URL: {progress.upload_url}")
+
+                # After iteration completes, the final response is available on the session
+                response = upload_session.response
+
+                # Handle the response
+                print(response)
+
+            def sample_upload_media_resume():
+                # Option 3: Resume an interrupted upload using a previously saved session URL.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request)
+
+                # Resume the interrupted upload from the saved session URL and chunk size
+                stream = io.BytesIO(b"Example upload data")
+                upload_url = "https://..."
+                chunk_size = 8 * 1024 * 1024
+                response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+
+                # Handle the response
+                print(response)
+
+            def sample_upload_media_resume_with_progress():
+                # Option 4: Resume an interrupted upload while receiving progress updates.
+                # Create a client
+                client = showcase_v1beta1.ResumableUploadServiceClient()
+
+                # Initialize request argument(s)
+                request = showcase_v1beta1.UploadMediaRequest(
+                )
+
+                # Create an upload session for the request
+                upload_session = client.upload_media(request=request)
+
+                # Resume the interrupted upload while iterating over progress updates
+                stream = io.BytesIO(b"Example upload data")
+                upload_url = "https://..."
+                chunk_size = 8 * 1024 * 1024
+                progress: UploadProgress
+                for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
+                    print(
+                        f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+                    )
+                    print(f"Session URL: {progress.upload_url}")
+                    print(f"Chunk size: {progress.chunk_size}")
+
+                # After iteration completes, the final response is available on the session
+                response = upload_session.response
 
                 # Handle the response
                 print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
index 39d286ead6d8..c4ce55a9da79 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
@@ -20,7 +20,7 @@
 # It may require modifications to work in your environment.
 
 # To install the latest published package dependency, execute the following:
-#   python3 -m pip install google-showcase
+#   python3 -m pip install google-showcase[async_rest]
 
 
 # [START localhost_v1beta1_generated_ResumableUploadService_UploadMedia_async]
@@ -32,9 +32,12 @@
 #   client as shown in:
 #   https://googleapis.dev/python/google-api-core/latest/client_options.html
 from google import showcase_v1beta1
+from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+import io
 
 
 async def sample_upload_media():
+    # Option 1: Upload a stream directly from start to completion.
     # Create a client
     client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
 
@@ -42,10 +45,110 @@ async def sample_upload_media():
     request = showcase_v1beta1.UploadMediaRequest(
     )
 
-    # Make the request
-    response = await client.upload_media(request=request)
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request, config=config)
+
+    # Upload the entire stream directly and await the final response
+    stream = io.BytesIO(b"Example upload data")
+    response = await upload_session.upload(stream)
+
+    # Handle the response
+    print(response)
+
+
+async def sample_upload_media_with_progress():
+    # Option 2: Upload a stream while receiving progress updates per chunk.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request, config=config)
+
+    # Iterate over the upload to receive progress updates as each chunk is transmitted
+    stream = io.BytesIO(b"Example upload data")
+    progress: UploadProgress
+    async for progress in upload_session.upload(stream):
+        print(
+            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+
+    # After iteration completes, the final response is available on the session
+    response = upload_session.response
 
     # Handle the response
     print(response)
 
+
+async def sample_upload_media_resume():
+    # Option 3: Resume an interrupted upload using a previously saved session URL.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request)
+
+    # Resume the interrupted upload from the saved session URL and chunk size
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+
+    # Handle the response
+    print(response)
+
+
+async def sample_upload_media_resume_with_progress():
+    # Option 4: Resume an interrupted upload while receiving progress updates.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request)
+
+    # Resume the interrupted upload while iterating over progress updates
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    progress: UploadProgress
+    async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
+        print(
+            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+        print(f"Chunk size: {progress.chunk_size}")
+
+    # After iteration completes, the final response is available on the session
+    response = upload_session.response
+
+    # Handle the response
+    print(response)
+
+
 # [END localhost_v1beta1_generated_ResumableUploadService_UploadMedia_async]
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
index 779d7aac1120..87e02ed921f7 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
@@ -32,9 +32,12 @@
 #   client as shown in:
 #   https://googleapis.dev/python/google-api-core/latest/client_options.html
 from google import showcase_v1beta1
+from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+import io
 
 
 def sample_upload_media():
+    # Option 1: Upload a stream directly from start to completion.
     # Create a client
     client = showcase_v1beta1.ResumableUploadServiceClient()
 
@@ -42,10 +45,110 @@ def sample_upload_media():
     request = showcase_v1beta1.UploadMediaRequest(
     )
 
-    # Make the request
-    response = client.upload_media(request=request)
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request, config=config)
+
+    # Upload the entire stream directly and return the final response
+    stream = io.BytesIO(b"Example upload data")
+    response = upload_session.upload(stream)
+
+    # Handle the response
+    print(response)
+
+
+def sample_upload_media_with_progress():
+    # Option 2: Upload a stream while receiving progress updates per chunk.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Configure optional transfer settings such as chunk size and stall detection
+    config = ResumableUploadConfig(
+        chunk_size=8 * 1024 * 1024,  # 8 MB
+        stall_minimum_rate=64 * 1024,
+        stall_timeout=120,
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request, config=config)
+
+    # Iterate over the upload to receive progress updates as each chunk is transmitted
+    stream = io.BytesIO(b"Example upload data")
+    progress: UploadProgress
+    for progress in upload_session.iter_upload(stream):
+        print(
+            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+
+    # After iteration completes, the final response is available on the session
+    response = upload_session.response
 
     # Handle the response
     print(response)
 
+
+def sample_upload_media_resume():
+    # Option 3: Resume an interrupted upload using a previously saved session URL.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request)
+
+    # Resume the interrupted upload from the saved session URL and chunk size
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+
+    # Handle the response
+    print(response)
+
+
+def sample_upload_media_resume_with_progress():
+    # Option 4: Resume an interrupted upload while receiving progress updates.
+    # Create a client
+    client = showcase_v1beta1.ResumableUploadServiceClient()
+
+    # Initialize request argument(s)
+    request = showcase_v1beta1.UploadMediaRequest(
+    )
+
+    # Create an upload session for the request
+    upload_session = client.upload_media(request=request)
+
+    # Resume the interrupted upload while iterating over progress updates
+    stream = io.BytesIO(b"Example upload data")
+    upload_url = "https://..."
+    chunk_size = 8 * 1024 * 1024
+    progress: UploadProgress
+    for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
+        print(
+            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
+        )
+        print(f"Session URL: {progress.upload_url}")
+        print(f"Chunk size: {progress.chunk_size}")
+
+    # After iteration completes, the final response is available on the session
+    response = upload_session.response
+
+    # Handle the response
+    print(response)
+
+
 # [END localhost_v1beta1_generated_ResumableUploadService_UploadMedia_sync]
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
index 6720f510f805..c87ef358f0e1 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
@@ -4812,33 +4812,31 @@
       "regionTag": "localhost_v1beta1_generated_ResumableUploadService_UploadMedia_async",
       "segments": [
         {
-          "end": 50,
+          "end": 153,
           "start": 27,
           "type": "FULL"
         },
         {
-          "end": 50,
+          "end": 153,
           "start": 27,
           "type": "SHORT"
         },
         {
-          "end": 40,
-          "start": 38,
+          "end": 127,
+          "start": 125,
           "type": "CLIENT_INITIALIZATION"
         },
         {
-          "end": 44,
-          "start": 41,
+          "start": 128,
           "type": "REQUEST_INITIALIZATION"
         },
         {
-          "end": 47,
-          "start": 45,
+          "end": 149,
           "type": "REQUEST_EXECUTION"
         },
         {
-          "end": 51,
-          "start": 48,
+          "end": 154,
+          "start": 150,
           "type": "RESPONSE_HANDLING"
         }
       ],
@@ -4888,33 +4886,31 @@
       "regionTag": "localhost_v1beta1_generated_ResumableUploadService_UploadMedia_sync",
       "segments": [
         {
-          "end": 50,
+          "end": 153,
           "start": 27,
           "type": "FULL"
         },
         {
-          "end": 50,
+          "end": 153,
           "start": 27,
           "type": "SHORT"
         },
         {
-          "end": 40,
-          "start": 38,
+          "end": 127,
+          "start": 125,
           "type": "CLIENT_INITIALIZATION"
         },
         {
-          "end": 44,
-          "start": 41,
+          "start": 128,
           "type": "REQUEST_INITIALIZATION"
         },
         {
-          "end": 47,
-          "start": 45,
+          "end": 149,
           "type": "REQUEST_EXECUTION"
         },
         {
-          "end": 51,
-          "start": 48,
+          "end": 154,
+          "start": 150,
           "type": "RESPONSE_HANDLING"
         }
       ],
diff --git a/packages/gapic-generator/tests/unit/samplegen/test_integration.py b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
index c937f78bddd9..3d15fa12a9d0 100644
--- a/packages/gapic-generator/tests/unit/samplegen/test_integration.py
+++ b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
@@ -807,3 +807,187 @@ def test_generate_sample_config_partial_config(fs):
 def test_generate_sample_config_fpaths_no_such_file(fs):
     with pytest.raises(types.InvalidConfig):
         list(gapic_utils.generate_all_sample_fpaths("cfgs/sample_config.yaml"))
+
+
+def test_generate_sample_resumable_upload():
+    classify_target_field = DummyField(
+        name="classify_target",
+        type=DummyMessageTypePB(name="ClassifyTarget"),
+        message=DummyMessage(
+            type="CLASSIFY TYPE",
+            fields={
+                "location_annotation": DummyField(
+                    type=DummyMessageTypePB(name="Location"),
+                    message=DummyMessage(type="LOCATION TYPE"),
+                ),
+            },
+        ),
+        ident=DummyIdent(sphinx="molluscs_v1.ClassifyTarget"),
+    )
+
+    input_type = DummyMessage(
+        type="REQUEST TYPE",
+        fields={"classify_target": classify_target_field},
+        ident=DummyIdent(
+            name="molluscs.v1.ClassifyRequest", sphinx="molluscs_v1.classify_request"
+        ),
+    )
+
+    output_type = DummyMessage(
+        type="RESPONSE TYPE",
+        fields={
+            "classification": DummyField(
+                type=DummyMessageTypePB(name="Classification"),
+            )
+        },
+        ident=DummyIdent(sphinx="molluscs_v1.classification"),
+    )
+
+    api_naming = naming.NewNaming(name="MolluscClient", namespace=("molluscs", "v1"))
+    service = wrappers.Service(
+        service_pb=namedtuple("service_pb", ["name"])("MolluscService"),
+        methods={
+            "Classify": DummyMethod(
+                name="Classify",
+                input=input_type,
+                output=message_factory("$resp.taxonomy"),
+                client_output_async=output_type,
+                client_output=output_type,
+                flattened_fields={"classify_target": classify_target_field},
+                is_resumable_upload=True,
+            )
+        },
+        visible_resources={},
+    )
+
+    schema = DummyApiSchema(
+        services={"animalia.mollusca.v1.Mollusc": service},
+        naming=api_naming,
+    )
+
+    sample = {
+        "service": "animalia.mollusca.v1.Mollusc",
+        "region_tag": "molluscs_generated_molluscs_v1_Mollusc_Classify_sync",
+        "rpc": "Classify",
+        "transport": "grpc",
+        "id": "mollusc_classify_sync",
+        "description": "Determine the full taxonomy of input mollusc",
+        "request": [
+            {
+                "field": "classify_target.location_annotation",
+                "value": "New Zealand",
+                "input_parameter": "location",
+            },
+        ],
+        "response": [{"print": ['Mollusc is a "%s"', "$resp.taxonomy"]}],
+    }
+
+    sample_str, _ = samplegen.generate_sample(
+        sample, schema, env.get_template("examples/sample.py.j2")
+    )
+
+    assert (
+        "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+        in sample_str
+    )
+    assert "import io" in sample_str
+    assert "upload_session = client.classify(request=request, config=config)" in sample_str
+    assert 'stream = io.BytesIO(b"Example upload data")' in sample_str
+    assert "response = upload_session.upload(stream)" in sample_str
+    assert "progress: UploadProgress" in sample_str
+    assert "for progress in upload_session.iter_upload(stream):" in sample_str
+    assert "response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)" in sample_str
+    assert "for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):" in sample_str
+
+
+def test_generate_sample_resumable_upload_async():
+    classify_target_field = DummyField(
+        name="classify_target",
+        type=DummyMessageTypePB(name="ClassifyTarget"),
+        message=DummyMessage(
+            type="CLASSIFY TYPE",
+            fields={
+                "location_annotation": DummyField(
+                    type=DummyMessageTypePB(name="Location"),
+                    message=DummyMessage(type="LOCATION TYPE"),
+                ),
+            },
+        ),
+        ident=DummyIdent(sphinx="molluscs_v1.ClassifyTarget"),
+    )
+
+    input_type = DummyMessage(
+        type="REQUEST TYPE",
+        fields={"classify_target": classify_target_field},
+        ident=DummyIdent(
+            name="molluscs.v1.ClassifyRequest", sphinx="molluscs_v1.classify_request"
+        ),
+    )
+
+    output_type = DummyMessage(
+        type="RESPONSE TYPE",
+        fields={
+            "classification": DummyField(
+                type=DummyMessageTypePB(name="Classification"),
+            )
+        },
+        ident=DummyIdent(sphinx="molluscs_v1.classification"),
+    )
+
+    api_naming = naming.NewNaming(name="MolluscClient", namespace=("molluscs", "v1"))
+    service = wrappers.Service(
+        service_pb=namedtuple("service_pb", ["name"])("MolluscService"),
+        methods={
+            "Classify": DummyMethod(
+                name="Classify",
+                input=input_type,
+                output=message_factory("$resp.taxonomy"),
+                client_output_async=output_type,
+                client_output=output_type,
+                flattened_fields={"classify_target": classify_target_field},
+                is_resumable_upload=True,
+            )
+        },
+        visible_resources={},
+    )
+
+    schema = DummyApiSchema(
+        services={"animalia.mollusca.v1.Mollusc": service},
+        naming=api_naming,
+    )
+
+    sample = {
+        "service": "animalia.mollusca.v1.Mollusc",
+        "region_tag": "molluscs_generated_molluscs_v1_Mollusc_Classify_async",
+        "rpc": "Classify",
+        "transport": "grpc-async",
+        "id": "mollusc_classify_async",
+        "description": "Determine the full taxonomy of input mollusc",
+        "request": [
+            {
+                "field": "classify_target.location_annotation",
+                "value": "New Zealand",
+                "input_parameter": "location",
+            },
+        ],
+        "response": [{"print": ['Mollusc is a "%s"', "$resp.taxonomy"]}],
+    }
+
+    sample_str, _ = samplegen.generate_sample(
+        sample, schema, env.get_template("examples/sample.py.j2")
+    )
+
+    assert (
+        "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+        in sample_str
+    )
+    assert "import io" in sample_str
+    assert "python3 -m pip install molluscs-v1-molluscclient[async_rest]" in sample_str
+    assert "upload_session = client.classify(request=request, config=config)" in sample_str
+    assert 'stream = io.BytesIO(b"Example upload data")' in sample_str
+    assert "response = await upload_session.upload(stream)" in sample_str
+    assert "progress: UploadProgress" in sample_str
+    assert "async for progress in upload_session.upload(stream):" in sample_str
+    assert "response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)" in sample_str
+    assert "async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):" in sample_str
+

From e2027fb75737bb697d1b6474ab1abbfa3f24ff52 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 5 Oct 2026 20:01:03 +0000
Subject: [PATCH 2/3] add missing await

---
 .../gapic/templates/examples/feature_fragments.j2               | 2 +-
 .../gapic-generator/tests/unit/samplegen/test_integration.py    | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2 b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
index 13d747e7a673..31f26ad39423 100644
--- a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
+++ b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
@@ -233,7 +233,7 @@ request=request
 {% macro render_method_call(sample, calling_form, calling_form_enum, transport, include_config=False) %}
 {# Note: this doesn't deal with enums or unions #}
 {# LROs return operation objects and paged requests return pager objects #}
-{% if transport == "grpc-async" and not sample.get("is_resumable_upload") and calling_form not in
+{% if transport == "grpc-async" and calling_form not in
 [calling_form_enum.RequestPaged, calling_form_enum.RequestPagedAll] %}
 await{{ " "}}
 {%- endif -%}
diff --git a/packages/gapic-generator/tests/unit/samplegen/test_integration.py b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
index 3d15fa12a9d0..84e26c7033e4 100644
--- a/packages/gapic-generator/tests/unit/samplegen/test_integration.py
+++ b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
@@ -983,7 +983,7 @@ def test_generate_sample_resumable_upload_async():
     )
     assert "import io" in sample_str
     assert "python3 -m pip install molluscs-v1-molluscclient[async_rest]" in sample_str
-    assert "upload_session = client.classify(request=request, config=config)" in sample_str
+    assert "upload_session = await client.classify(request=request, config=config)" in sample_str
     assert 'stream = io.BytesIO(b"Example upload data")' in sample_str
     assert "response = await upload_session.upload(stream)" in sample_str
     assert "progress: UploadProgress" in sample_str

From d3a287006d6319442317a37a8511804a0938ef80 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 5 Oct 2026 22:11:07 +0000
Subject: [PATCH 3/3] address review feedback

---
 .../gapic/samplegen/samplegen.py              |   2 +-
 .../templates/examples/feature_fragments.j2   | 101 +++--------------
 .../resumable_upload_service/async_client.py  | 102 +++--------------
 .../resumable_upload_service/client.py        | 100 +++--------------
 ...mable_upload_service_upload_media_async.py | 105 +++---------------
 ...umable_upload_service_upload_media_sync.py | 103 +++--------------
 ...ppet_metadata_google.showcase.v1beta1.json |  36 +++---
 .../tests/unit/samplegen/test_integration.py  |  21 ++--
 8 files changed, 105 insertions(+), 465 deletions(-)

diff --git a/packages/gapic-generator/gapic/samplegen/samplegen.py b/packages/gapic-generator/gapic/samplegen/samplegen.py
index cca1bf0b0a87..690117d0d975 100644
--- a/packages/gapic-generator/gapic/samplegen/samplegen.py
+++ b/packages/gapic-generator/gapic/samplegen/samplegen.py
@@ -1192,7 +1192,7 @@ def _get_sample_imports(sample: Dict, rpc: wrappers.Method) -> List[str]:
 
     if rpc.is_resumable_upload:
         imports.append(
-            "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+            "from google.api_core.resumable_transfer import ResumableUploadConfig"
         )
         imports.append("import io")
 
diff --git a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2 b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
index 31f26ad39423..5a995725c6cb 100644
--- a/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
+++ b/packages/gapic-generator/gapic/templates/examples/feature_fragments.j2
@@ -254,9 +254,7 @@ client.{{ render_method_name(sample)|trim }}({{ render_request_params_unary(samp
 {% set fn_name = sample.rpc|snake_case|trim %}
 {% set fn_params = print_input_params(sample.request)|trim %}
 {% set method_call_with_config = render_method_call(sample, calling_form, calling_form_enum, sample.transport, include_config=True)|trim %}
-{% set method_call_no_config = render_method_call(sample, calling_form, calling_form_enum, sample.transport, include_config=False)|trim %}
 {{ async_prefix }}def sample_{{ fn_name }}({{ fn_params }}):
-    # Option 1: Upload a stream directly from start to completion.
     {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
     {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
     # Configure optional transfer settings such as chunk size and stall detection
@@ -266,106 +264,37 @@ client.{{ render_method_name(sample)|trim }}({{ render_request_params_unary(samp
         stall_timeout=120,
     )
 
-    # Create an upload session for the request
+    # Make the request
     upload_session = {{ method_call_with_config }}
 
-    # Upload the entire stream directly and {% if is_async %}await{% else %}return{% endif %} the final response
+    # Option 1: Upload the entire stream directly and {% if is_async %}await{% else %}return{% endif %} the final response
     stream = io.BytesIO(b"Example upload data")
     {{ "response = " if sample.response else "" }}{{ await_prefix }}upload_session.upload(stream)
-    {% if sample.response %}
-
-    # Handle the response
-    {% for statement in sample.response %}
-    {{ dispatch_statement(statement)|trim }}
-    {% endfor %}
-    {% endif %}
-
-
-{{ async_prefix }}def sample_{{ fn_name }}_with_progress({{ fn_params }}):
-    # Option 2: Upload a stream while receiving progress updates per chunk.
-    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
-    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
-    # Configure optional transfer settings such as chunk size and stall detection
-    config = ResumableUploadConfig(
-        chunk_size=8 * 1024 * 1024,  # 8 MB
-        stall_minimum_rate=64 * 1024,
-        stall_timeout=120,
-    )
 
-    # Create an upload session for the request
-    upload_session = {{ method_call_with_config }}
-
-    # Iterate over the upload to receive progress updates as each chunk is transmitted
-    stream = io.BytesIO(b"Example upload data")
-    progress: UploadProgress
+    # Option 2: Alternatively, iterate over the upload to receive progress updates per chunk
     {% if is_async %}
-    async for progress in upload_session.upload(stream):
+    # async for progress in upload_session.upload(stream):
     {% else %}
-    for progress in upload_session.iter_upload(stream):
+    # for progress in upload_session.iter_upload(stream):
     {% endif %}
-        print(
-            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-
-    # After iteration completes, the final response is available on the session
-    {{ "response = " if sample.response else "" }}upload_session.response
+    #     print(f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+    #     print(f"Session URL: {progress.upload_url}")
     {% if sample.response %}
-
-    # Handle the response
-    {% for statement in sample.response %}
-    {{ dispatch_statement(statement)|trim }}
-    {% endfor %}
+    # response = upload_session.response
     {% endif %}
 
+    # Option 3: Alternatively, resume an interrupted upload from a saved session URL
+    # {{ "response = " if sample.response else "" }}{{ await_prefix }}upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)
 
-{{ async_prefix }}def sample_{{ fn_name }}_resume({{ fn_params }}):
-    # Option 3: Resume an interrupted upload using a previously saved session URL.
-    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
-    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
-    # Create an upload session for the request
-    upload_session = {{ method_call_no_config }}
-
-    # Resume the interrupted upload from the saved session URL and chunk size
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    {{ "response = " if sample.response else "" }}{{ await_prefix }}upload_session.resume(upload_url, stream, chunk_size=chunk_size)
-    {% if sample.response %}
-
-    # Handle the response
-    {% for statement in sample.response %}
-    {{ dispatch_statement(statement)|trim }}
-    {% endfor %}
-    {% endif %}
-
-
-{{ async_prefix }}def sample_{{ fn_name }}_resume_with_progress({{ fn_params }}):
-    # Option 4: Resume an interrupted upload while receiving progress updates.
-    {{ render_client_setup(sample.module_name, sample.client_name)|indent }}
-    {{ render_request_setup(sample.request, sample.request_module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
-    # Create an upload session for the request
-    upload_session = {{ method_call_no_config }}
-
-    # Resume the interrupted upload while iterating over progress updates
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    progress: UploadProgress
+    # Option 4: Alternatively, resume an interrupted upload while receiving progress updates
     {% if is_async %}
-    async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
+    # async for progress in upload_session.resume(upload_url, stream, chunk_size=config.chunk_size):
     {% else %}
-    for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
+    # for progress in upload_session.iter_resume(upload_url, stream, chunk_size=config.chunk_size):
     {% endif %}
-        print(
-            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-        print(f"Chunk size: {progress.chunk_size}")
-
-    # After iteration completes, the final response is available on the session
-    {{ "response = " if sample.response else "" }}upload_session.response
+    #     print(f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}")
     {% if sample.response %}
+    # response = upload_session.response
 
     # Handle the response
     {% for statement in sample.response %}
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
index b3d88f76894a..adb806e016af 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/async_client.py
@@ -280,11 +280,10 @@ async def upload_media(self,
             #   client as shown in:
             #   https://googleapis.dev/python/google-api-core/latest/client_options.html
             from google import showcase_v1beta1
-            from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+            from google.api_core.resumable_transfer import ResumableUploadConfig
             import io
 
             async def sample_upload_media():
-                # Option 1: Upload a stream directly from start to completion.
                 # Create a client
                 client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
 
@@ -299,97 +298,26 @@ async def sample_upload_media():
                     stall_timeout=120,
                 )
 
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request, config=config)
+                # Make the request
+                upload_session = await client.upload_media(request=request, config=config)
 
-                # Upload the entire stream directly and await the final response
+                # Option 1: Upload the entire stream directly and await the final response
                 stream = io.BytesIO(b"Example upload data")
                 response = await upload_session.upload(stream)
 
-                # Handle the response
-                print(response)
-
-            async def sample_upload_media_with_progress():
-                # Option 2: Upload a stream while receiving progress updates per chunk.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
-
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
-
-                # Configure optional transfer settings such as chunk size and stall detection
-                config = ResumableUploadConfig(
-                    chunk_size=8 * 1024 * 1024,  # 8 MB
-                    stall_minimum_rate=64 * 1024,
-                    stall_timeout=120,
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request, config=config)
-
-                # Iterate over the upload to receive progress updates as each chunk is transmitted
-                stream = io.BytesIO(b"Example upload data")
-                progress: UploadProgress
-                async for progress in upload_session.upload(stream):
-                    print(
-                        f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-                    )
-                    print(f"Session URL: {progress.upload_url}")
-
-                # After iteration completes, the final response is available on the session
-                response = upload_session.response
-
-                # Handle the response
-                print(response)
-
-            async def sample_upload_media_resume():
-                # Option 3: Resume an interrupted upload using a previously saved session URL.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
-
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request)
-
-                # Resume the interrupted upload from the saved session URL and chunk size
-                stream = io.BytesIO(b"Example upload data")
-                upload_url = "https://..."
-                chunk_size = 8 * 1024 * 1024
-                response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)
+                # Option 2: Alternatively, iterate over the upload to receive progress updates per chunk
+                # async for progress in upload_session.upload(stream):
+                #     print(f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+                #     print(f"Session URL: {progress.upload_url}")
+                # response = upload_session.response
 
-                # Handle the response
-                print(response)
-
-            async def sample_upload_media_resume_with_progress():
-                # Option 4: Resume an interrupted upload while receiving progress updates.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
+                # Option 3: Alternatively, resume an interrupted upload from a saved session URL
+                # response = await upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)
 
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request)
-
-                # Resume the interrupted upload while iterating over progress updates
-                stream = io.BytesIO(b"Example upload data")
-                upload_url = "https://..."
-                chunk_size = 8 * 1024 * 1024
-                progress: UploadProgress
-                async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
-                    print(
-                        f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-                    )
-                    print(f"Session URL: {progress.upload_url}")
-                    print(f"Chunk size: {progress.chunk_size}")
-
-                # After iteration completes, the final response is available on the session
-                response = upload_session.response
+                # Option 4: Alternatively, resume an interrupted upload while receiving progress updates
+                # async for progress in upload_session.resume(upload_url, stream, chunk_size=config.chunk_size):
+                #     print(f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+                # response = upload_session.response
 
                 # Handle the response
                 print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
index 2626bd662a3e..1f62d75b9951 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/client.py
@@ -514,11 +514,10 @@ def upload_media(self,
             #   client as shown in:
             #   https://googleapis.dev/python/google-api-core/latest/client_options.html
             from google import showcase_v1beta1
-            from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+            from google.api_core.resumable_transfer import ResumableUploadConfig
             import io
 
             def sample_upload_media():
-                # Option 1: Upload a stream directly from start to completion.
                 # Create a client
                 client = showcase_v1beta1.ResumableUploadServiceClient()
 
@@ -533,97 +532,26 @@ def sample_upload_media():
                     stall_timeout=120,
                 )
 
-                # Create an upload session for the request
+                # Make the request
                 upload_session = client.upload_media(request=request, config=config)
 
-                # Upload the entire stream directly and return the final response
+                # Option 1: Upload the entire stream directly and return the final response
                 stream = io.BytesIO(b"Example upload data")
                 response = upload_session.upload(stream)
 
-                # Handle the response
-                print(response)
-
-            def sample_upload_media_with_progress():
-                # Option 2: Upload a stream while receiving progress updates per chunk.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceClient()
-
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
+                # Option 2: Alternatively, iterate over the upload to receive progress updates per chunk
+                # for progress in upload_session.iter_upload(stream):
+                #     print(f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+                #     print(f"Session URL: {progress.upload_url}")
+                # response = upload_session.response
 
-                # Configure optional transfer settings such as chunk size and stall detection
-                config = ResumableUploadConfig(
-                    chunk_size=8 * 1024 * 1024,  # 8 MB
-                    stall_minimum_rate=64 * 1024,
-                    stall_timeout=120,
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request, config=config)
-
-                # Iterate over the upload to receive progress updates as each chunk is transmitted
-                stream = io.BytesIO(b"Example upload data")
-                progress: UploadProgress
-                for progress in upload_session.iter_upload(stream):
-                    print(
-                        f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-                    )
-                    print(f"Session URL: {progress.upload_url}")
-
-                # After iteration completes, the final response is available on the session
-                response = upload_session.response
-
-                # Handle the response
-                print(response)
+                # Option 3: Alternatively, resume an interrupted upload from a saved session URL
+                # response = upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)
 
-            def sample_upload_media_resume():
-                # Option 3: Resume an interrupted upload using a previously saved session URL.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceClient()
-
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request)
-
-                # Resume the interrupted upload from the saved session URL and chunk size
-                stream = io.BytesIO(b"Example upload data")
-                upload_url = "https://..."
-                chunk_size = 8 * 1024 * 1024
-                response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)
-
-                # Handle the response
-                print(response)
-
-            def sample_upload_media_resume_with_progress():
-                # Option 4: Resume an interrupted upload while receiving progress updates.
-                # Create a client
-                client = showcase_v1beta1.ResumableUploadServiceClient()
-
-                # Initialize request argument(s)
-                request = showcase_v1beta1.UploadMediaRequest(
-                )
-
-                # Create an upload session for the request
-                upload_session = client.upload_media(request=request)
-
-                # Resume the interrupted upload while iterating over progress updates
-                stream = io.BytesIO(b"Example upload data")
-                upload_url = "https://..."
-                chunk_size = 8 * 1024 * 1024
-                progress: UploadProgress
-                for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
-                    print(
-                        f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-                    )
-                    print(f"Session URL: {progress.upload_url}")
-                    print(f"Chunk size: {progress.chunk_size}")
-
-                # After iteration completes, the final response is available on the session
-                response = upload_session.response
+                # Option 4: Alternatively, resume an interrupted upload while receiving progress updates
+                # for progress in upload_session.iter_resume(upload_url, stream, chunk_size=config.chunk_size):
+                #     print(f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+                # response = upload_session.response
 
                 # Handle the response
                 print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
index c4ce55a9da79..2bc21047931c 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_async.py
@@ -32,12 +32,11 @@
 #   client as shown in:
 #   https://googleapis.dev/python/google-api-core/latest/client_options.html
 from google import showcase_v1beta1
-from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+from google.api_core.resumable_transfer import ResumableUploadConfig
 import io
 
 
 async def sample_upload_media():
-    # Option 1: Upload a stream directly from start to completion.
     # Create a client
     client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
 
@@ -52,100 +51,26 @@ async def sample_upload_media():
         stall_timeout=120,
     )
 
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request, config=config)
+    # Make the request
+    upload_session = await client.upload_media(request=request, config=config)
 
-    # Upload the entire stream directly and await the final response
+    # Option 1: Upload the entire stream directly and await the final response
     stream = io.BytesIO(b"Example upload data")
     response = await upload_session.upload(stream)
 
-    # Handle the response
-    print(response)
-
-
-async def sample_upload_media_with_progress():
-    # Option 2: Upload a stream while receiving progress updates per chunk.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Configure optional transfer settings such as chunk size and stall detection
-    config = ResumableUploadConfig(
-        chunk_size=8 * 1024 * 1024,  # 8 MB
-        stall_minimum_rate=64 * 1024,
-        stall_timeout=120,
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request, config=config)
-
-    # Iterate over the upload to receive progress updates as each chunk is transmitted
-    stream = io.BytesIO(b"Example upload data")
-    progress: UploadProgress
-    async for progress in upload_session.upload(stream):
-        print(
-            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-
-    # After iteration completes, the final response is available on the session
-    response = upload_session.response
-
-    # Handle the response
-    print(response)
+    # Option 2: Alternatively, iterate over the upload to receive progress updates per chunk
+    # async for progress in upload_session.upload(stream):
+    #     print(f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+    #     print(f"Session URL: {progress.upload_url}")
+    # response = upload_session.response
 
+    # Option 3: Alternatively, resume an interrupted upload from a saved session URL
+    # response = await upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)
 
-async def sample_upload_media_resume():
-    # Option 3: Resume an interrupted upload using a previously saved session URL.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request)
-
-    # Resume the interrupted upload from the saved session URL and chunk size
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)
-
-    # Handle the response
-    print(response)
-
-
-async def sample_upload_media_resume_with_progress():
-    # Option 4: Resume an interrupted upload while receiving progress updates.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceAsyncClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request)
-
-    # Resume the interrupted upload while iterating over progress updates
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    progress: UploadProgress
-    async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):
-        print(
-            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-        print(f"Chunk size: {progress.chunk_size}")
-
-    # After iteration completes, the final response is available on the session
-    response = upload_session.response
+    # Option 4: Alternatively, resume an interrupted upload while receiving progress updates
+    # async for progress in upload_session.resume(upload_url, stream, chunk_size=config.chunk_size):
+    #     print(f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+    # response = upload_session.response
 
     # Handle the response
     print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
index 87e02ed921f7..b1165a305eed 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/localhost_v1beta1_generated_resumable_upload_service_upload_media_sync.py
@@ -32,12 +32,11 @@
 #   client as shown in:
 #   https://googleapis.dev/python/google-api-core/latest/client_options.html
 from google import showcase_v1beta1
-from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress
+from google.api_core.resumable_transfer import ResumableUploadConfig
 import io
 
 
 def sample_upload_media():
-    # Option 1: Upload a stream directly from start to completion.
     # Create a client
     client = showcase_v1beta1.ResumableUploadServiceClient()
 
@@ -52,100 +51,26 @@ def sample_upload_media():
         stall_timeout=120,
     )
 
-    # Create an upload session for the request
+    # Make the request
     upload_session = client.upload_media(request=request, config=config)
 
-    # Upload the entire stream directly and return the final response
+    # Option 1: Upload the entire stream directly and return the final response
     stream = io.BytesIO(b"Example upload data")
     response = upload_session.upload(stream)
 
-    # Handle the response
-    print(response)
-
-
-def sample_upload_media_with_progress():
-    # Option 2: Upload a stream while receiving progress updates per chunk.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Configure optional transfer settings such as chunk size and stall detection
-    config = ResumableUploadConfig(
-        chunk_size=8 * 1024 * 1024,  # 8 MB
-        stall_minimum_rate=64 * 1024,
-        stall_timeout=120,
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request, config=config)
-
-    # Iterate over the upload to receive progress updates as each chunk is transmitted
-    stream = io.BytesIO(b"Example upload data")
-    progress: UploadProgress
-    for progress in upload_session.iter_upload(stream):
-        print(
-            f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-
-    # After iteration completes, the final response is available on the session
-    response = upload_session.response
-
-    # Handle the response
-    print(response)
+    # Option 2: Alternatively, iterate over the upload to receive progress updates per chunk
+    # for progress in upload_session.iter_upload(stream):
+    #     print(f"Uploaded {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+    #     print(f"Session URL: {progress.upload_url}")
+    # response = upload_session.response
 
+    # Option 3: Alternatively, resume an interrupted upload from a saved session URL
+    # response = upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)
 
-def sample_upload_media_resume():
-    # Option 3: Resume an interrupted upload using a previously saved session URL.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request)
-
-    # Resume the interrupted upload from the saved session URL and chunk size
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)
-
-    # Handle the response
-    print(response)
-
-
-def sample_upload_media_resume_with_progress():
-    # Option 4: Resume an interrupted upload while receiving progress updates.
-    # Create a client
-    client = showcase_v1beta1.ResumableUploadServiceClient()
-
-    # Initialize request argument(s)
-    request = showcase_v1beta1.UploadMediaRequest(
-    )
-
-    # Create an upload session for the request
-    upload_session = client.upload_media(request=request)
-
-    # Resume the interrupted upload while iterating over progress updates
-    stream = io.BytesIO(b"Example upload data")
-    upload_url = "https://..."
-    chunk_size = 8 * 1024 * 1024
-    progress: UploadProgress
-    for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):
-        print(
-            f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}"
-        )
-        print(f"Session URL: {progress.upload_url}")
-        print(f"Chunk size: {progress.chunk_size}")
-
-    # After iteration completes, the final response is available on the session
-    response = upload_session.response
+    # Option 4: Alternatively, resume an interrupted upload while receiving progress updates
+    # for progress in upload_session.iter_resume(upload_url, stream, chunk_size=config.chunk_size):
+    #     print(f"Resumed {progress.bytes_uploaded} bytes | State: {progress.state.name}")
+    # response = upload_session.response
 
     # Handle the response
     print(response)
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
index c87ef358f0e1..c64d375bfa25 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/samples/generated_samples/snippet_metadata_google.showcase.v1beta1.json
@@ -4812,31 +4812,33 @@
       "regionTag": "localhost_v1beta1_generated_ResumableUploadService_UploadMedia_async",
       "segments": [
         {
-          "end": 153,
+          "end": 78,
           "start": 27,
           "type": "FULL"
         },
         {
-          "end": 153,
+          "end": 78,
           "start": 27,
           "type": "SHORT"
         },
         {
-          "end": 127,
-          "start": 125,
+          "end": 42,
+          "start": 40,
           "type": "CLIENT_INITIALIZATION"
         },
         {
-          "start": 128,
+          "end": 53,
+          "start": 43,
           "type": "REQUEST_INITIALIZATION"
         },
         {
-          "end": 149,
+          "end": 74,
+          "start": 54,
           "type": "REQUEST_EXECUTION"
         },
         {
-          "end": 154,
-          "start": 150,
+          "end": 79,
+          "start": 75,
           "type": "RESPONSE_HANDLING"
         }
       ],
@@ -4886,31 +4888,33 @@
       "regionTag": "localhost_v1beta1_generated_ResumableUploadService_UploadMedia_sync",
       "segments": [
         {
-          "end": 153,
+          "end": 78,
           "start": 27,
           "type": "FULL"
         },
         {
-          "end": 153,
+          "end": 78,
           "start": 27,
           "type": "SHORT"
         },
         {
-          "end": 127,
-          "start": 125,
+          "end": 42,
+          "start": 40,
           "type": "CLIENT_INITIALIZATION"
         },
         {
-          "start": 128,
+          "end": 53,
+          "start": 43,
           "type": "REQUEST_INITIALIZATION"
         },
         {
-          "end": 149,
+          "end": 74,
+          "start": 54,
           "type": "REQUEST_EXECUTION"
         },
         {
-          "end": 154,
-          "start": 150,
+          "end": 79,
+          "start": 75,
           "type": "RESPONSE_HANDLING"
         }
       ],
diff --git a/packages/gapic-generator/tests/unit/samplegen/test_integration.py b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
index 84e26c7033e4..65fa11acf144 100644
--- a/packages/gapic-generator/tests/unit/samplegen/test_integration.py
+++ b/packages/gapic-generator/tests/unit/samplegen/test_integration.py
@@ -887,17 +887,17 @@ def test_generate_sample_resumable_upload():
     )
 
     assert (
-        "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+        "from google.api_core.resumable_transfer import ResumableUploadConfig"
         in sample_str
     )
     assert "import io" in sample_str
+    assert "# Make the request" in sample_str
     assert "upload_session = client.classify(request=request, config=config)" in sample_str
     assert 'stream = io.BytesIO(b"Example upload data")' in sample_str
     assert "response = upload_session.upload(stream)" in sample_str
-    assert "progress: UploadProgress" in sample_str
-    assert "for progress in upload_session.iter_upload(stream):" in sample_str
-    assert "response = upload_session.resume(upload_url, stream, chunk_size=chunk_size)" in sample_str
-    assert "for progress in upload_session.iter_resume(upload_url, stream, chunk_size=chunk_size):" in sample_str
+    assert "# for progress in upload_session.iter_upload(stream):" in sample_str
+    assert "# response = upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)" in sample_str
+    assert "# for progress in upload_session.iter_resume(upload_url, stream, chunk_size=config.chunk_size):" in sample_str
 
 
 def test_generate_sample_resumable_upload_async():
@@ -978,16 +978,17 @@ def test_generate_sample_resumable_upload_async():
     )
 
     assert (
-        "from google.api_core.resumable_transfer import ResumableUploadConfig, UploadProgress"
+        "from google.api_core.resumable_transfer import ResumableUploadConfig"
         in sample_str
     )
     assert "import io" in sample_str
     assert "python3 -m pip install molluscs-v1-molluscclient[async_rest]" in sample_str
+    assert "# Make the request" in sample_str
     assert "upload_session = await client.classify(request=request, config=config)" in sample_str
     assert 'stream = io.BytesIO(b"Example upload data")' in sample_str
     assert "response = await upload_session.upload(stream)" in sample_str
-    assert "progress: UploadProgress" in sample_str
-    assert "async for progress in upload_session.upload(stream):" in sample_str
-    assert "response = await upload_session.resume(upload_url, stream, chunk_size=chunk_size)" in sample_str
-    assert "async for progress in upload_session.resume(upload_url, stream, chunk_size=chunk_size):" in sample_str
+    assert "# async for progress in upload_session.upload(stream):" in sample_str
+    assert "# response = await upload_session.resume(upload_url, stream, chunk_size=config.chunk_size)" in sample_str
+    assert "# async for progress in upload_session.resume(upload_url, stream, chunk_size=config.chunk_size):" in sample_str
+