feat(images): support GCS references and legacy BlobKey fallback for custom gRPC redirection

- Increase gRPC max receive message size to 32MB to handle large processed images.
- Bypass blobstore.create_gs_key when USE_CUSTOM_IMAGES_GRPC_SERVICE is active, passing raw GCS paths directly to the service.
- Implement local blob content fetching in _set_imagedata for legacy BlobKeys. This allows the custom gRPC service (running on Cloud Run) to receive the image data as bytes since it cannot resolve proprietary BlobKeys.
- Update imports to support authenticated gRPC and high-level blob operations.

Author: abhishekr700

src/google/appengine/api/images/__init__.py

@@ -37,14 +37,15 @@
 import os
 import struct
 
-# import google.auth
-# import google.auth.transport.grpc
-# import google.auth.transport.requests
+import google.auth
+import google.auth.transport.grpc
+import google.auth.transport.requests
 from google.oauth2 import id_token
 import grpc
 from google.appengine.api import apiproxy_stub_map
 from google.appengine.api import blobstore
 from google.appengine.api import datastore_types
+from google.appengine.ext import blobstore as ext_blobstore
 from . import images_service_pb2
 from . import images_service_rpc_pb2_grpc
 from google.appengine.runtime import apiproxy_errors
@@ -88,7 +89,10 @@ def _make_grpc_call(method_name, request):
         call_credentials,
     )
 
-    authed_channel = grpc.secure_channel(target_host, composite_credentials)
+    # Increase max receive message size to 32MB to handle large images
+    options = [('grpc.max_receive_message_length', 32 * 1024 * 1024)]
+    
+    authed_channel = grpc.secure_channel(target_host, composite_credentials, options=options)
     stub = images_service_rpc_pb2_grpc.ImagesServiceStub(authed_channel)
 
     grpc_method = getattr(stub, method_name)
@@ -266,7 +270,10 @@ def __init__(self, image_data=None, blob_key=None, filename=None):
 
     self._image_data = image_data
     if filename:
-      self._blob_key = blobstore.create_gs_key(filename)
+      if _USE_GRPC:
+        self._blob_key = filename
+      else:
+        self._blob_key = blobstore.create_gs_key(filename)
     else:
       self._blob_key = _extract_blob_key(blob_key)
     self._transforms = []
@@ -814,8 +821,29 @@ def _set_imagedata(self, imagedata):
         imagedata: An `ImageData` protocol buffer instance.
     """
     if self._blob_key:
-      imagedata.content = b""
-      imagedata.blob_key = self._blob_key
+      if _USE_GRPC and not self._blob_key.startswith('/gs/'):
+        # If we are using the custom gRPC service and have a legacy BlobKey,
+        # we must fetch the data locally because the external service cannot read it.
+        if not self._image_data:
+          try:
+            # Use high-level ext.blobstore to fetch content
+            blob_info = ext_blobstore.BlobInfo.get(self._blob_key)
+            if not blob_info:
+               raise ObjectNotFoundError()
+            
+            with blob_info.open() as reader:
+              self._image_data = reader.read()
+          except Exception as e:
+            if isinstance(e, ObjectNotFoundError):
+               raise
+            raise Error("Failed to fetch blob data for gRPC: %s" % e)
+        
+        imagedata.content = self._image_data
+      else:
+        # For GCS paths (starting with /gs/) or standard App Engine calls,
+        # we maintain the reference-based approach.
+        imagedata.content = b""
+        imagedata.blob_key = self._blob_key
     else:
       imagedata.content = self._image_data