Gemini API × Cloud Storage: Building a Production Pipeline for 100MB File Processing with Private DB Integration

Setup and context — The File Size Barrier Has Been Lifted

One of the biggest bottlenecks when using the Gemini API in production environments has been the 20MB file upload limit. Long meeting recordings, multi-hundred-page technical documents, high-resolution product image catalogs — real-world business files routinely exceed 20MB, making chunk splitting and compression a mandatory preprocessing step.

In April 2026, Google raised this limit to 100MB and simultaneously introduced native support for Cloud Storage (GCS) buckets and private DB pre-signed URLs as data input sources. This means you no longer need to download and re-upload files — you can point Gemini directly at data sitting in GCS.

This article provides a systematic guide to designing and implementing production-grade file processing pipelines that leverage these new capabilities.

Target audience:

Developers struggling with large file processing in Gemini API
Backend engineers building integration pipelines with GCS or S3
Tech leads running multimodal AI in production environments

For the fundamentals of Gemini API's multimodal capabilities, see our guide on building a multimodal document analysis system with Gemini API. This article builds on those foundations with a focus on Cloud Storage integration and large file handling.

Cloud Storage Integration Overview

Three Input Patterns

The new file input capabilities for Gemini API fall into three main patterns.

Pattern 1: Direct GCS Bucket Reference

You can specify files stored in Google Cloud Storage directly using gs:// URIs. This eliminates the download-then-reupload cycle, dramatically reducing network transfer and latency.

Pattern 2: Pre-signed URLs

Beyond GCS, you can pass pre-signed URLs from AWS S3, Azure Blob Storage, or any private database. This is ideal for multi-cloud environments and on-premises data source integration.

Pattern 3: Traditional File Upload (Now Up to 100MB)

The Files API direct upload limit has been expanded to 100MB. This remains effective for small-to-medium files or architectures that don't use cloud storage.

Supported File Formats and Limits

The 100MB limit applies to all existing multimodal file formats:

Documents: PDF (hundreds of pages), text files
Images: JPEG, PNG, WebP, GIF
Audio: MP3, WAV, FLAC, OGG (approximately 9.5 hours)
Video: MP4, MOV, MPEG (approximately 1 hour)

Implementing GCS Bucket Integration

Basic Implementation with Python SDK

import google.generativeai as genai
from google.cloud import storage
import os
 
# Configure API key
genai.configure(api_key=os.environ["YOUR_GEMINI_API_KEY"])
 
# Reference files directly from GCS bucket
model = genai.GenerativeModel("gemini-2.5-pro")
 
# Specify GCS URI to process the file directly
response = model.generate_content([
    "Summarize the key points of this technical document in a structured format. "
    "Rate the importance of each section on a 5-point scale and note implementation caveats.",
    {"file_uri": "gs://my-project-bucket/docs/technical-spec-v3.pdf",
     "mime_type": "application/pdf"}
])
 
print(response.text)
# Example output:
# ## Technical Document Summary
# ### Section 1: Architecture Overview (Importance: ★★★★★)
# - Microservices architecture with...
# ### Section 2: API Design (Importance: ★★★★☆)
# ...

Private DB Integration with Pre-signed URLs

import google.generativeai as genai
from google.cloud import storage
from datetime import timedelta
 
def generate_signed_url(bucket_name: str, blob_name: str, 
                         expiration_minutes: int = 15) -> str:
    """Generate a GCS signed URL"""
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    
    url = blob.generate_signed_url(
        version="v4",
        expiration=timedelta(minutes=expiration_minutes),
        method="GET"
    )
    return url
 
# Process private files via signed URL
signed_url = generate_signed_url(
    "my-private-bucket", 
    "reports/quarterly-analysis-q1-2026.pdf"
)
 
model = genai.GenerativeModel("gemini-2.5-pro")
response = model.generate_content([
    "Extract revenue trends and risk factors from this quarterly report.",
    {"file_uri": signed_url,
     "mime_type": "application/pdf"}
])
 
print(response.text)
# Example output:
# ## Q1 2026 Quarterly Analysis
# ### Revenue Trends
# - Year-over-year growth of +23%...
# ### Identified Risk Factors
# 1. Supply chain delay risks...

Cross-Cloud Integration with AWS S3

In multi-cloud environments, S3 pre-signed URLs work just as seamlessly.

import boto3
from botocore.config import Config
import google.generativeai as genai
 
def get_s3_presigned_url(bucket: str, key: str, 
                          expiration: int = 900) -> str:
    """Generate an AWS S3 pre-signed URL"""
    s3_client = boto3.client(
        's3',
        config=Config(signature_version='s3v4'),
        region_name='us-east-1'
    )
    url = s3_client.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=expiration
    )
    return url
 
# Analyze a video file from S3 with Gemini
s3_url = get_s3_presigned_url(
    "my-media-bucket",
    "recordings/meeting-2026-04-01.mp4"
)
 
model = genai.GenerativeModel("gemini-2.5-pro")
response = model.generate_content([
    "Create meeting minutes from this recording. "
    "Organize by speaker and extract action items.",
    {"file_uri": s3_url,
     "mime_type": "video/mp4"}
])
 
print(response.text)

Designing a Batch File Processing Pipeline

Architecture Overview

Processing large volumes of files efficiently requires pipeline architecture rather than individual API calls. Here's a production pipeline combining GCS, Cloud Functions, and Gemini API.

Processing flow:

Files are uploaded to a GCS bucket
Cloud Functions (or Cloud Run) triggers on the event
File metadata validation (size and format checks)
Gemini API processes the file via GCS URI
Results are stored in Firestore/BigQuery
Failures are routed to a Dead Letter Queue

Event-Driven Processing with Cloud Functions

import functions_framework
import google.generativeai as genai
from google.cloud import firestore
import json
import logging
 
logger = logging.getLogger(__name__)
 
# Initialize Gemini model at module level (cold start optimization)
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-flash")
db = firestore.Client()
 
# File extension to MIME type mapping
MIME_MAP = {
    ".pdf": "application/pdf",
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".mp4": "video/mp4",
    ".mp3": "audio/mpeg",
    ".wav": "audio/wav",
}
 
@functions_framework.cloud_event
def process_uploaded_file(cloud_event):
    """Auto-triggered file processing function on GCS upload events"""
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]
    file_size = int(data.get("size", 0))
    
    # Validation
    if file_size > 100 * 1024 * 1024:  # 100MB
        logger.error(f"File too large: {file_name} ({file_size} bytes)")
        return {"status": "error", "reason": "file_too_large"}
    
    # Determine MIME type from extension
    ext = "." + file_name.rsplit(".", 1)[-1].lower()
    mime_type = MIME_MAP.get(ext)
    if not mime_type:
        logger.warning(f"Unsupported file type: {ext}")
        return {"status": "skipped", "reason": "unsupported_type"}
    
    # Build GCS URI
    gcs_uri = f"gs://{bucket_name}/{file_name}"
    
    try:
        # Process with Gemini API
        response = model.generate_content([
            "Analyze this file and return results in the following JSON format: "
            '{"summary": "overview", "key_points": ["point1", "point2"], '
            '"category": "category", "language": "detected language"}',
            {"file_uri": gcs_uri, "mime_type": mime_type}
        ])
        
        # Save results to Firestore
        result = json.loads(response.text)
        doc_ref = db.collection("processed_files").document()
        doc_ref.set({
            "file_uri": gcs_uri,
            "file_name": file_name,
            "file_size": file_size,
            "mime_type": mime_type,
            "analysis": result,
            "processed_at": firestore.SERVER_TIMESTAMP,
            "status": "completed"
        })
        
        logger.info(f"Successfully processed: {file_name}")
        return {"status": "success", "doc_id": doc_ref.id}
        
    except Exception as e:
        logger.error(f"Processing failed for {file_name}: {e}")
        # Route to Dead Letter Queue
        db.collection("processing_errors").add({
            "file_uri": gcs_uri,
            "error": str(e),
            "timestamp": firestore.SERVER_TIMESTAMP,
            "retry_count": 0
        })
        return {"status": "error", "reason": str(e)}

Error Handling and Retry Strategies

Error Patterns Unique to Large Files

Processing 100MB files introduces error patterns you won't encounter with smaller files.

Timeout errors: Large PDFs and long videos take significantly longer to analyze. The default timeout (60 seconds) is often insufficient and needs to be adjusted.

Rate limiting: Large files consume more tokens, making it easier to hit RPM/TPM limits.

Partial processing failures: When part of a file is corrupted, the API may return results for only the analyzable portions rather than failing entirely.

Retry Implementation with Exponential Backoff

import time
import random
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
 
def process_with_retry(model, contents, max_retries=5, 
                        base_delay=2.0, max_delay=60.0):
    """Retry processing with Exponential Backoff + Jitter"""
    
    for attempt in range(max_retries):
        try:
            response = model.generate_content(
                contents,
                request_options={"timeout": 300}  # 5-minute timeout
            )
            return response
            
        except google_exceptions.ResourceExhausted as e:
            # Rate limited — always retry
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            wait_time = delay + jitter
            print(f"Rate limited (attempt {attempt+1}/{max_retries}). "
                  f"Waiting {wait_time:.1f}s...")
            time.sleep(wait_time)
            
        except google_exceptions.DeadlineExceeded as e:
            # Timeout — file may be too large
            if attempt < max_retries - 1:
                print(f"Timeout (attempt {attempt+1}). "
                      f"Consider using a faster model or smaller file.")
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise
                
        except google_exceptions.InvalidArgument as e:
            # File format errors — no retry needed
            raise
    
    raise Exception(f"Max retries ({max_retries}) exceeded")
 
# Usage example
model = genai.GenerativeModel("gemini-2.5-pro")
response = process_with_retry(model, [
    "Summarize this 100-page PDF by chapter.",
    {"file_uri": "gs://my-bucket/large-document.pdf",
     "mime_type": "application/pdf"}
])

Cost Optimization Strategies

Cost Control Through Model Selection

Large file processing consumes significant tokens, making model selection directly impact costs. For a comprehensive guide to Gemini API cost optimization strategies, see our complete cost optimization guide.

Gemini 2.5 Flash: Fast and cost-effective. Ideal for routine file classification, metadata extraction, and simple summarization
Gemini 2.5 Pro: High accuracy. Use for complex analysis, multilingual document processing, and detailed report generation
Flex Tier: For non-real-time batch processing, the Flex tier can reduce costs by up to 50%

Two-Stage Processing Pattern

The best practice for cost optimization is a two-stage processing approach.

import google.generativeai as genai
 
# Stage 1: Fast triage with Flash (low cost)
flash_model = genai.GenerativeModel("gemini-2.5-flash")
 
triage_response = flash_model.generate_content([
    "Analyze this file and determine the following: "
    '{"complexity": "low/medium/high", "language": "language", '
    '"estimated_pages": number, "requires_deep_analysis": true/false}',
    {"file_uri": "gs://my-bucket/report.pdf",
     "mime_type": "application/pdf"}
])
 
import json
triage = json.loads(triage_response.text)
 
# Stage 2: Deep analysis with Pro only for complex files
if triage.get("requires_deep_analysis"):
    pro_model = genai.GenerativeModel("gemini-2.5-pro")
    detailed_response = pro_model.generate_content([
        "Create a detailed analysis report for this document. "
        "Comprehensively extract legal risks, financial considerations, "
        "and technical challenges.",
        {"file_uri": "gs://my-bucket/report.pdf",
         "mime_type": "application/pdf"}
    ])
    print(detailed_response.text)
else:
    # Flash results are sufficient
    print(f"Simple file — Flash result used. Language: {triage['language']}")

Combining with Context Caching

When asking multiple questions about the same file, Context Caching dramatically reduces token costs for cache-hit portions.

import google.generativeai as genai
from google.generativeai import caching
import datetime
 
# Create a cache (requires at least 32,768 tokens of content)
cache = caching.CachedContent.create(
    model="models/gemini-2.5-flash",
    display_name="quarterly-report-q1",
    contents=[
        {"file_uri": "gs://my-bucket/q1-report-2026.pdf",
         "mime_type": "application/pdf"}
    ],
    ttl=datetime.timedelta(hours=2)  # Valid for 2 hours
)
 
# Run multiple queries at reduced cost using the cache
cached_model = genai.GenerativeModel.from_cached_content(cache)
 
questions = [
    "What is the year-over-year revenue growth rate?",
    "Which business area carries the highest risk?",
    "What external factors will impact next quarter's forecast?"
]
 
for q in questions:
    response = cached_model.generate_content(q)
    print(f"Q: {q}")
    print(f"A: {response.text}\n")
 
# Delete cache when done
cache.delete()

Security and Access Control

Pre-signed URL Best Practices

Here are the key security considerations when working with pre-signed URLs.

Set the shortest possible expiration: Configure the minimum time needed for file processing (typically 15–30 minutes) to minimize unnecessary access risk.

Principle of least privilege for IAM roles: Grant only storage.objects.get to the Cloud Functions service account and isolate write permissions separately.

VPC Service Controls: In production environments, configure VPC Service Controls to restrict Gemini API access to authorized networks only.

# Example: Generating a secure signed URL
from google.cloud import storage
from datetime import timedelta
 
def create_secure_signed_url(bucket_name: str, blob_name: str) -> str:
    """Generate a least-privilege signed URL"""
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    
    url = blob.generate_signed_url(
        version="v4",
        expiration=timedelta(minutes=15),  # Shortest viable expiration
        method="GET",
        # Restrict content type
        response_type="application/pdf",
    )
    return url

Production Monitoring

Collecting Processing Metrics

Stable operation of large file pipelines requires continuous monitoring of key metrics.

import time
from google.cloud import monitoring_v3
from google.protobuf import timestamp_pb2
 
class PipelineMetrics:
    """Collect and send pipeline metrics"""
    
    def __init__(self, project_id: str):
        self.client = monitoring_v3.MetricServiceClient()
        self.project_name = f"projects/{project_id}"
    
    def record_processing(self, file_size_mb: float, 
                           processing_time_s: float,
                           model: str, success: bool):
        """Record processing result metrics"""
        series = monitoring_v3.TimeSeries()
        series.metric.type = "custom.googleapis.com/gemini/file_processing"
        series.metric.labels["model"] = model
        series.metric.labels["success"] = str(success).lower()
        series.resource.type = "global"
        
        point = monitoring_v3.Point()
        point.value.double_value = processing_time_s
        now = time.time()
        point.interval.end_time.seconds = int(now)
        series.points.append(point)
        
        self.client.create_time_series(
            name=self.project_name,
            time_series=[series]
        )
 
# Usage example
metrics = PipelineMetrics("my-gcp-project")
 
start = time.time()
response = model.generate_content([...])
elapsed = time.time() - start
 
metrics.record_processing(
    file_size_mb=85.2,
    processing_time_s=elapsed,
    model="gemini-2.5-flash",
    success=True
)

Summary

The Gemini API's Cloud Storage integration and 100MB file support represent a significant leap in the practicality of multimodal AI pipelines. Direct GCS URI references, cross-cloud integration via pre-signed URLs, two-stage processing for cost optimization, and event-driven batch processing — combining these patterns lets you build simple yet robust pipelines for large file AI processing that previously required extensive preprocessing effort.

The combination of Flex Tier and Context Caching is particularly powerful for dramatically reducing document processing costs at scale. We recommend starting with a small prototype and incrementally scaling to production.

To deepen your understanding of the pipeline architecture covered in this article,