From 97f5a696cdb3629472abfe25fd8d4471e759c761 Mon Sep 17 00:00:00 2001
From: Vasilii Novikov <novikov.vz@gmail.com>
Date: Thu, 12 Jun 2025 12:24:04 +0200
Subject: [PATCH 1/2] feat: make GCS artifact service async

---
 .../adk/artifacts/gcs_artifact_service.py     | 123 +++++++++++++++---
 1 file changed, 104 insertions(+), 19 deletions(-)

diff --git a/src/google/adk/artifacts/gcs_artifact_service.py b/src/google/adk/artifacts/gcs_artifact_service.py
index 6a700a1d480..dc80f98d6ff 100644
--- a/src/google/adk/artifacts/gcs_artifact_service.py
+++ b/src/google/adk/artifacts/gcs_artifact_service.py
@@ -22,6 +22,8 @@
 """
 from __future__ import annotations
 
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
 import logging
 from typing import Optional
 
@@ -37,16 +39,106 @@
 class GcsArtifactService(BaseArtifactService):
   """An artifact service implementation using Google Cloud Storage (GCS)."""
 
-  def __init__(self, bucket_name: str, **kwargs):
+  def __init__(self, bucket_name: str, thread_pool_executor=None, **kwargs):
     """Initializes the GcsArtifactService.
 
+
     Args:
         bucket_name: The name of the bucket to use.
+        thread_pool_executor: `concurrent.futures.ThreadPoolExecutor`, an instance of executor to achieve concurrency
+          on top of synchronous GCS client.
+          If not specified, the default executor will be used.
         **kwargs: Keyword arguments to pass to the Google Cloud Storage client.
     """
     self.bucket_name = bucket_name
     self.storage_client = storage.Client(**kwargs)
     self.bucket = self.storage_client.bucket(self.bucket_name)
+    if not thread_pool_executor:
+      thread_pool_executor = ThreadPoolExecutor()
+    self._executor = thread_pool_executor
+
+  @override
+  async def save_artifact(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      session_id: str,
+      filename: str,
+      artifact: types.Part,
+  ) -> int:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        self._executor,
+        self._save_artifact,
+        app_name,
+        user_id,
+        session_id,
+        filename,
+        artifact,
+    )
+
+  @override
+  async def load_artifact(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      session_id: str,
+      filename: str,
+      version: Optional[int] = None,
+  ) -> Optional[types.Part]:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        self._executor,
+        self._load_artifact,
+        app_name,
+        user_id,
+        session_id,
+        filename,
+        version,
+    )
+
+  @override
+  async def list_artifact_keys(
+      self, *, app_name: str, user_id: str, session_id: str
+  ) -> list[str]:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        self._executor,
+        self._list_artifact_keys,
+        app_name,
+        user_id,
+        session_id,
+    )
+
+  @override
+  async def delete_artifact(
+      self, *, app_name: str, user_id: str, session_id: str, filename: str
+  ) -> None:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        self._executor,
+        self._delete_artifact,
+        app_name,
+        user_id,
+        session_id,
+        filename,
+    )
+
+  @override
+  async def list_versions(
+      self, *, app_name: str, user_id: str, session_id: str, filename: str
+  ) -> list[int]:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        self._executor,
+        self._list_versions,
+        app_name,
+        user_id,
+        session_id,
+        filename,
+    )
 
   def _file_has_user_namespace(self, filename: str) -> bool:
     """Checks if the filename has a user namespace.
@@ -84,17 +176,15 @@ def _get_blob_name(
       return f"{app_name}/{user_id}/user/{filename}/{version}"
     return f"{app_name}/{user_id}/{session_id}/{filename}/{version}"
 
-  @override
-  async def save_artifact(
+  def _save_artifact(
       self,
-      *,
       app_name: str,
       user_id: str,
       session_id: str,
       filename: str,
       artifact: types.Part,
   ) -> int:
-    versions = await self.list_versions(
+    versions = self._list_versions(
         app_name=app_name,
         user_id=user_id,
         session_id=session_id,
@@ -114,10 +204,8 @@ async def save_artifact(
 
     return version
 
-  @override
-  async def load_artifact(
+  def _load_artifact(
       self,
-      *,
       app_name: str,
       user_id: str,
       session_id: str,
@@ -125,7 +213,7 @@ async def load_artifact(
       version: Optional[int] = None,
   ) -> Optional[types.Part]:
     if version is None:
-      versions = await self.list_versions(
+      versions = self._list_versions(
           app_name=app_name,
           user_id=user_id,
           session_id=session_id,
@@ -148,9 +236,8 @@ async def load_artifact(
     )
     return artifact
 
-  @override
-  async def list_artifact_keys(
-      self, *, app_name: str, user_id: str, session_id: str
+  def _list_artifact_keys(
+      self, app_name: str, user_id: str, session_id: str
   ) -> list[str]:
     filenames = set()
 
@@ -172,11 +259,10 @@ async def list_artifact_keys(
 
     return sorted(list(filenames))
 
-  @override
-  async def delete_artifact(
-      self, *, app_name: str, user_id: str, session_id: str, filename: str
+  def _delete_artifact(
+      self, app_name: str, user_id: str, session_id: str, filename: str
   ) -> None:
-    versions = await self.list_versions(
+    versions = self._list_versions(
         app_name=app_name,
         user_id=user_id,
         session_id=session_id,
@@ -190,9 +276,8 @@ async def delete_artifact(
       blob.delete()
     return
 
-  @override
-  async def list_versions(
-      self, *, app_name: str, user_id: str, session_id: str, filename: str
+  def _list_versions(
+      self, app_name: str, user_id: str, session_id: str, filename: str
   ) -> list[int]:
     """Lists all available versions of an artifact.
 

From 64d872a38d98741e638cc99d430c3d3051caca70 Mon Sep 17 00:00:00 2001
From: Vasilii Novikov <novikov.vz@gmail.com>
Date: Sat, 19 Jul 2025 13:33:26 +0300
Subject: [PATCH 2/2] Use asyncio.to_thread instead of thread pool executor

---
 .../adk/artifacts/gcs_artifact_service.py     | 29 ++++---------------
 1 file changed, 6 insertions(+), 23 deletions(-)

diff --git a/src/google/adk/artifacts/gcs_artifact_service.py b/src/google/adk/artifacts/gcs_artifact_service.py
index dc80f98d6ff..ff6a17b6474 100644
--- a/src/google/adk/artifacts/gcs_artifact_service.py
+++ b/src/google/adk/artifacts/gcs_artifact_service.py
@@ -23,7 +23,6 @@
 from __future__ import annotations
 
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 import logging
 from typing import Optional
 
@@ -39,23 +38,17 @@
 class GcsArtifactService(BaseArtifactService):
   """An artifact service implementation using Google Cloud Storage (GCS)."""
 
-  def __init__(self, bucket_name: str, thread_pool_executor=None, **kwargs):
+  def __init__(self, bucket_name: str, **kwargs):
     """Initializes the GcsArtifactService.
 
 
     Args:
         bucket_name: The name of the bucket to use.
-        thread_pool_executor: `concurrent.futures.ThreadPoolExecutor`, an instance of executor to achieve concurrency
-          on top of synchronous GCS client.
-          If not specified, the default executor will be used.
         **kwargs: Keyword arguments to pass to the Google Cloud Storage client.
     """
     self.bucket_name = bucket_name
     self.storage_client = storage.Client(**kwargs)
     self.bucket = self.storage_client.bucket(self.bucket_name)
-    if not thread_pool_executor:
-      thread_pool_executor = ThreadPoolExecutor()
-    self._executor = thread_pool_executor
 
   @override
   async def save_artifact(
@@ -67,9 +60,7 @@ async def save_artifact(
       filename: str,
       artifact: types.Part,
   ) -> int:
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(
-        self._executor,
+    return await asyncio.to_thread(
         self._save_artifact,
         app_name,
         user_id,
@@ -88,9 +79,7 @@ async def load_artifact(
       filename: str,
       version: Optional[int] = None,
   ) -> Optional[types.Part]:
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(
-        self._executor,
+    return await asyncio.to_thread(
         self._load_artifact,
         app_name,
         user_id,
@@ -103,9 +92,7 @@ async def load_artifact(
   async def list_artifact_keys(
       self, *, app_name: str, user_id: str, session_id: str
   ) -> list[str]:
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(
-        self._executor,
+    return await asyncio.to_thread(
         self._list_artifact_keys,
         app_name,
         user_id,
@@ -116,9 +103,7 @@ async def list_artifact_keys(
   async def delete_artifact(
       self, *, app_name: str, user_id: str, session_id: str, filename: str
   ) -> None:
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(
-        self._executor,
+    return await asyncio.to_thread(
         self._delete_artifact,
         app_name,
         user_id,
@@ -130,9 +115,7 @@ async def delete_artifact(
   async def list_versions(
       self, *, app_name: str, user_id: str, session_id: str, filename: str
   ) -> list[int]:
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(
-        self._executor,
+    return await asyncio.to_thread(
         self._list_versions,
         app_name,
         user_id,