fix: Remove bytes for Gemini Developer API before sending a video for Veo Video Extension async

PiperOrigin-RevId: 822280664

Author: matthew29tang

google/genai/models.py

@@ -7277,6 +7277,36 @@ async def generate_videos(
           'Source and prompt/image/video are mutually exclusive.'
           + ' Please only use source.'
       )
+    # Gemini Developer API does not support video bytes.
+    video_dct: dict[str, Any] = {}
+    if not self._api_client.vertexai and video:
+      if isinstance(video, types.Video):
+        video_dct = video.model_dump()
+      else:
+        video_dct = dict(video)
+
+      if video_dct.get('uri') and video_dct.get('video_bytes'):
+        video = types.Video(
+            uri=video_dct.get('uri'), mime_type=video_dct.get('mime_type')
+        )
+    elif not self._api_client.vertexai and source:
+      if isinstance(source, types.GenerateVideosSource):
+        source_dct = source.model_dump()
+        video_dct = source_dct.get('video', {})
+      else:
+        source_dct = dict(source)
+        if isinstance(source_dct.get('video'), types.Video):
+          video_obj: types.Video = source_dct.get('video', types.Video())
+          video_dct = video_obj.model_dump()
+      if video_dct and video_dct.get('uri') and video_dct.get('video_bytes'):
+        source = types.GenerateVideosSource(
+            prompt=source_dct.get('prompt'),
+            image=source_dct.get('image'),
+            video=types.Video(
+                uri=video_dct.get('uri'),
+                mime_type=video_dct.get('mime_type'),
+            ),
+        )
     return await self._generate_videos(
         model=model,
         prompt=prompt,

google/genai/tests/models/test_generate_videos.py

@@ -700,3 +700,48 @@ async def test_text_to_video_poll_async(client):
     operation = await client.aio.operations.get(operation=operation)
 
   assert operation.result.generated_videos[0].video.uri
+
+
+@pytest.mark.asyncio
+async def test_generated_video_extension_from_source_poll_async(client):
+  # Gemini API only supports video extension on generated videos.
+  if client.vertexai:
+    return
+
+  operation1 = await client.aio.models.generate_videos(
+      model="veo-3.1-generate-preview",
+      prompt="Rain",
+      config=types.GenerateVideosConfig(
+          number_of_videos=1,
+      ),
+  )
+  while not operation1.done:
+    # Skip the sleep when in replay mode.
+    if client._api_client._mode not in ("replay", "auto"):
+      time.sleep(20)
+    operation1 = await client.aio.operations.get(operation=operation1)
+
+  video1 = operation1.result.generated_videos[0].video
+  assert video1.uri
+  assert await client.aio.files.download(file=video1)
+
+  operation2 = await client.aio.models.generate_videos(
+      model="veo-3.1-generate-preview",
+      source=types.GenerateVideosSource(
+          prompt="Sun",
+          video=video1
+      ),
+      config=types.GenerateVideosConfig(
+          number_of_videos=1,
+      ),
+  )
+  while not operation2.done:
+    # Skip the sleep when in replay mode.
+    if client._api_client._mode not in ("replay", "auto"):
+      time.sleep(20)
+    operation2 = await client.aio.operations.get(operation=operation2)
+
+  video2 = operation2.result.generated_videos[0].video
+  assert video2.uri
+  assert await client.aio.files.download(file=video2)
+