Being Able to Say Later Where a User's Data Was Processed — Region-Pinned Gemini (Vertex AI) Clients and a Residency Policy Gate

Insert a region-pinned client

Let us look at the form that is easy to write by accident, then replace it with a pinned version.

# Before: default client. Nothing specifies where processing happens
from google import genai
 
client = genai.Client()  # Developer API default, roughly global
 
def summarize(text: str) -> str:
    resp = client.models.generate_content(
        model="gemini-flash-latest",
        contents=text,
    )
    return resp.text

This works, but the concept of a region is nowhere in the code. We switch to Vertex AI and supply a client with an explicit location, created once per class and reused.

# After: resolve class -> region and supply a location-pinned client
from google import genai
from functools import lru_cache
 
PROJECT = "my-gcp-project"
 
# Per-class "first-choice region" and "allowed regions"
REGION_POLICY = {
    "residency":     {"primary": "europe-west4", "allowed": {"europe-west4", "europe-west1"}, "global_ok": False},
    "regional-pref": {"primary": "europe-west4", "allowed": {"europe-west4", "europe-west1"}, "global_ok": False},
    "global-ok":     {"primary": "global",       "allowed": {"global"},                       "global_ok": True},
}
 
@lru_cache(maxsize=8)
def _client_for(location: str) -> genai.Client:
    # Create and cache exactly one client per location
    return genai.Client(vertexai=True, project=PROJECT, location=location)
 
def client_for_class(data_class: str, location: str | None = None) -> genai.Client:
    policy = REGION_POLICY[data_class]
    loc = location or policy["primary"]
    if loc not in policy["allowed"]:
        raise ResidencyError(f"{data_class} does not allow region {loc}")
    return _client_for(loc)

The location argument of genai.Client(vertexai=True, project=..., location=...) determines which regional endpoint you actually hit. Pinning it per class turns "residency data is always called from europe-west4" from a setting into a guaranteed path through the code. The lru_cache avoids rebuilding a client for every call to the same region.

A policy gate that forbids implicit global fallback

The most dangerous thing in residency is the single line "on error, just retry on global." Fallback for availability is added with good intentions, but doing it for the residency class breaks exactly the property you set out to protect. The gate must be fail-closed—when in doubt, stop.

class ResidencyError(RuntimeError):
    """Stop calls that could violate a residency constraint"""
 
def generate(data_class: str, *, model: str, contents, location: str | None = None):
    policy = REGION_POLICY[data_class]
    client = client_for_class(data_class, location)
    used_location = location or policy["primary"]
    try:
        resp = client.models.generate_content(model=model, contents=contents)
        _ledger_write(data_class, used_location, model, status="ok")
        return resp
    except Exception as e:
        # Do not escape to global here. Retry only within allowed regions
        for alt in policy["allowed"]:
            if alt == used_location:
                continue
            try:
                resp = _client_for(alt).models.generate_content(model=model, contents=contents)
                _ledger_write(data_class, alt, model, status="ok-failover")
                return resp
            except Exception:
                continue
        _ledger_write(data_class, used_location, model, status="failed")
        if not policy["global_ok"]:
            # residency/regional-pref: return a failure rather than dropping to global
            raise ResidencyError(f"{data_class}: could not process within allowed regions") from e
        # Only global-ok may take a final fallback to global here
        resp = _client_for("global").models.generate_content(model=model, contents=contents)
        _ledger_write(data_class, "global", model, status="ok-global-fallback")
        return resp

The key is that the fallback search is confined to policy["allowed"]. Availability is secured "within allowed regions," and if that still fails, the residency class fails honestly. Returning an error so a human can decide is the correct design in this context—far better than swallowing the failure and answering from global.

Turn "where it was processed" into something you can state later

Even with pinning in code, without records you cannot later show that processing stayed in the EU. For each call, write the class, the region actually used, the model, and the outcome to a lightweight ledger. SQLite is plenty.

import sqlite3, time, uuid
from pathlib import Path
 
_DB = Path("data/residency_ledger.db")
 
def _ledger_init():
    con = sqlite3.connect(_DB)
    con.execute("""
      CREATE TABLE IF NOT EXISTS calls(
        id TEXT PRIMARY KEY, ts REAL, data_class TEXT,
        location TEXT, model TEXT, status TEXT
      )""")
    con.commit(); con.close()
 
def _ledger_write(data_class: str, location: str, model: str, status: str):
    con = sqlite3.connect(_DB)
    con.execute(
        "INSERT INTO calls VALUES(?,?,?,?,?,?)",
        (uuid.uuid4().hex, time.time(), data_class, location, model, status),
    )
    con.commit(); con.close()

With this ledger you can answer, in a single query, "how many residency-class calls happened last month, and were any of them processed outside europe-west4?"

SELECT location, status, COUNT(*) AS n
FROM calls
WHERE data_class = 'residency'
  AND ts >= strftime('%s', '2026-06-01')
GROUP BY location, status
ORDER BY n DESC;

If even one residency row shows global or ok-global-fallback, that is proof the design leaked. Ideally, results for this class are filled only with allowed regions. The ledger is both the evidence that you held the line and the alerting source that catches the moment it broke. I run this query daily and get notified if a residency row lands in a disallowed region.

Measure what region pinning costs in latency

Pinning has a price. You should measure how much in your own environment before deciding. For my EU-facing summarization task (short text, roughly gemini-flash-latest), running 500 calls each on global and on a europe-west4 pin gave this distribution.

For EU users, pinning improved both p50 and p95. I attribute the longer tail on global to the requests that were routed somewhere far. The pinned side, however, introduced a different kind of failure: "the model I want is not in this region." The newest preview models can arrive later in some regions. That is the next issue.

What to do when the model is not in that region

The conflict "I want to hold residency, but the latest model I want is not yet in europe-west4" happens in practice. Escaping to global here would make the original policy meaningless. The order I use is this.

First, for the residency class I keep an explicit second choice: "a one-step-back stable model that runs in an allowed region." The judgment is to prioritize honoring the class constraint over chasing the newest model. Second, if nothing at all runs in an allowed region, that feature is temporarily not offered to residency users (fail-closed). Third, I record the count of these "could not serve" cases in the ledger and restore the feature once regional availability catches up.

RESIDENCY_MODEL_FALLBACK = {
    # Stable model to degrade to when the first choice is absent in an allowed region
    "gemini-flash-latest": "gemini-2.5-flash",
}
 
def pick_model(data_class: str, wanted: str, location: str) -> str:
    if data_class == "residency" and not _model_available(wanted, location):
        alt = RESIDENCY_MODEL_FALLBACK.get(wanted)
        if alt and _model_available(alt, location):
            return alt
        raise ResidencyError(f"neither {wanted} nor a fallback is available in {location}")
    return wanted

In practice, _model_available should fetch the available models per region once at startup and cache them. Querying on every call becomes latency and cost in itself.

A way to start small

You do not need to swap everything at once. I added the pieces in this order.

First, add a layer that only tags the data class at the entry point. At this stage everything can still be treated as global-ok. Behavior does not change, so it is safe to introduce. Next, enable only the ledger writes and observe for about a week what is being processed and where. Only then does your app's actual data flow become visible. With that in hand, raise a single, clearly-residency entry point to the residency class and let the policy gate take effect. Confirm it runs well, then widen the scope.

The nice thing about this order is that at every stage you repeat "observe, tighten exactly one place, confirm." A constraint like residency, if you chase perfection all at once, tends to stall you with availability drops and missing models. Watching reality through the ledger and pinning entry points in order of sensitivity ends up faster and more stable.

Wrap-up: your next move

If you are calling Gemini on global today, before changing much code, just run the call ledger for a week. You will see, in numbers, which entry points correspond to residency in your app and how often they are hit. Then pin only the single most sensitive entry point to something like europe-west4 and cut off the implicit fallback to global. Taking just those two steps reliably moves you out of the state where you cannot answer where processing happened.