Metadata-Version: 2.4
Name: gemini-gateway
Version: 0.3.1
Summary: Shared Google Gemini API gateway with multi-key rate limiting and retries.
Author-email: NickZaitsev <nekitlomic@gmail.com>
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/NickZaitsev/gemini-gateway
Project-URL: Repository, https://github.com/NickZaitsev/gemini-gateway
Project-URL: Issues, https://github.com/NickZaitsev/gemini-gateway/issues
Keywords: gemini,google-genai,rate-limit,llm,api-gateway
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: google-genai>=1.30.0
Requires-Dist: pydantic>=2.8.0
Provides-Extra: dotenv
Requires-Dist: python-dotenv>=1.0.1; extra == "dotenv"
Provides-Extra: dev
Requires-Dist: build>=1.2.2; extra == "dev"
Requires-Dist: pytest>=8.3.2; extra == "dev"
Requires-Dist: ruff>=0.8.0; extra == "dev"
Requires-Dist: pyright>=1.1.390; extra == "dev"
Requires-Dist: twine>=5.1.1; extra == "dev"
Dynamic: license-file

# gemini-gateway

[![CI](https://github.com/NickZaitsev/gemini-gateway/actions/workflows/ci.yml/badge.svg)](https://github.com/NickZaitsev/gemini-gateway/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/gemini-gateway)](https://pypi.org/project/gemini-gateway/)
[![Python](https://img.shields.io/pypi/pyversions/gemini-gateway)](https://pypi.org/project/gemini-gateway/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](LICENSE)

A small, dependency-light Python gateway for the Google Gemini API that adds the
production concerns you usually end up writing yourself: **multi-key rotation,
client-side rate limiting, bounded retries, error classification, and typed
structured output** — behind a tiny, stable API.

It is deliberately scoped to reusable API plumbing. It contains no
application-specific logic, so the same gateway can be shared across projects by
giving each one its own environment-variable prefix.

---

## Table of contents

- [Why](#why)
- [Features](#features)
- [Requirements](#requirements)
- [Installation](#installation)
- [Quick start](#quick-start)
- [Structured JSON output](#structured-json-output)
- [Result metadata](#result-metadata)
- [Configuration](#configuration)
  - [Environment variables](#environment-variables)
  - [Per-project prefixes](#per-project-prefixes)
  - [Configuring in code](#configuring-in-code)
- [How rate limiting works](#how-rate-limiting-works)
- [Retries and error handling](#retries-and-error-handling)
- [Testing your own code](#testing-your-own-code)
- [Public API](#public-api)
- [Development](#development)
- [License](#license)

---

## Why

Calling Gemini directly works until it doesn't: a key hits its per-minute quota,
the backend returns a transient `503`, or you want JSON back as a validated
object instead of a string you have to parse by hand. `gemini-gateway` wraps the
official [`google-genai`](https://pypi.org/project/google-genai/) client and
handles those cases consistently, without leaking provider details into your
application code.

## Features

- **Multi-key rotation** — supply several API keys; requests are spread across
  them and a key is automatically skipped while it is cooling down or disabled.
- **Client-side rate limiting** — enforces requests-per-minute (RPM),
  tokens-per-minute (TPM), and requests-per-day (RPD) limits *before* sending,
  so you stay under quota instead of reacting to `429`s.
- **Bounded retries with backoff** — transient failures are retried with
  exponential backoff and honor the API's `retryDelay` hint when present. Retry
  counts are bounded by default; unbounded retries on capacity errors are strict
  opt-in.
- **Error classification** — exceptions are categorized (`rate_limit`,
  `network`, `server`, `auth`, `validation`, `internal`, `unknown`) using
  structured status codes first, with message-text fallback.
- **Typed structured output** — pass a Pydantic model and get a validated
  instance back.
- **Usage metadata** — optional result objects expose input/output token counts,
  the model used, and which key served the request.
- **Testable by design** — the network client, clock, and sleep function are all
  injectable, so your tests never touch the real API.

## Requirements

- Python **3.10+**
- A Google Gemini API key ([get one here](https://aistudio.google.com/apikey))

## Installation

```bash
pip install gemini-gateway
```

## Quick start

```python
from gemini_gateway import GeminiGateway

# Reads GEMINI_API_KEY / GEMINI_API_KEYS (and other GEMINI_* vars) from the
# environment or a local .env file.
gateway = GeminiGateway.from_env()

text = gateway.generate_text("Write a one-sentence product tagline for a coffee shop.")
print(text)
```

The minimum configuration is a single API key:

```bash
export GEMINI_API_KEY="your-key"
```

## Structured JSON output

Pass a Pydantic model and receive a validated instance. The gateway requests
JSON from Gemini using the model as the response schema and validates the result
for you.

```python
from pydantic import BaseModel
from gemini_gateway import GeminiGateway


class Product(BaseModel):
    name: str
    tagline: str
    price_usd: float


gateway = GeminiGateway.from_env()

product = gateway.generate_json(
    "Invent a fictional coffee product as JSON with name, tagline, price_usd.",
    Product,
    max_output_tokens=512,
)

print(product.name, product.price_usd)  # fully typed
```

If Gemini returns malformed or non-conforming JSON, a Pydantic
`ValidationError` (a subclass of `ValueError`) is raised.

## Result metadata

Use the `*_result` variants when you need token usage, the model name, or which
key served the request (useful for logging and cost tracking).

```python
result = gateway.generate_text_result("Summarize the theory of relativity.")

print(result.text)
print(result.usage.input_tokens, result.usage.output_tokens)
print(result.model)  # e.g. "gemini-3.1-flash-lite"
print(result.api_key_label)  # e.g. "key-2"
```

`generate_json_result(...)` returns the same metadata with a validated
`payload` field instead of `text`.

## Configuration

### Environment variables

All variables are prefixed (default prefix: `GEMINI`). Only an API key is
required; everything else has a sensible default.

| Variable (with prefix)                | Default            | Description                                                        |
| ------------------------------------- | ------------------ | ------------------------------------------------------------------ |
| `_API_KEYS`                           | —                  | Comma/newline/semicolon-separated list of keys.                    |
| `_API_KEY`                            | —                  | Single key (used if `_API_KEYS` is unset).                         |
| `_MODEL`                              | `gemini-3.1-flash-lite` | Primary model. May itself be a comma-separated list: the first entry is primary, the rest become fallbacks. |
| `_FALLBACK_MODELS`                    | —                  | Comma-separated fallback models, tried only when the primary is exhausted. |
| `_PROXY_URL`                          | —                  | Optional HTTP(S) proxy URL used only by Gemini API clients.        |
| `_RPM`                                | `15`               | Max requests per minute, per key.                                  |
| `_TPM`                                | `250000`           | Max tokens per minute, per key.                                    |
| `_RPD`                                | `500`              | Max requests per day, per key (rolling 24h window).                |
| `_TIMEOUT_MS`                         | `60000`            | Per-request HTTP timeout, in milliseconds.                         |
| `_MAX_RETRIES`                        | `3`                | Max attempts for retryable errors.                                 |
| `_MAX_OUTPUT_TOKENS`                  | `2048`             | Default output token cap (overridable per call).                   |
| `_TEMPERATURE`                        | `0.6`              | Sampling temperature (0–2).                                        |
| `_RETRY_BASE_SECONDS`                 | `2.0`              | Base delay for exponential backoff.                                |
| `_RETRY_MAX_SECONDS`                  | `60.0`             | Maximum backoff delay.                                             |
| `_DEFAULT_COOLDOWN_SECONDS`           | `5.0`              | Cooldown applied to a key after a limit/error with no hint.        |
| `_DAILY_QUOTA_COOLDOWN_SECONDS`       | `10800.0`          | How long a `(key, model)` pair is parked after a per-day project quota. |
| `_RESPONSE_SCHEMA_MODE`               | `pydantic`         | `pydantic` sends the model type as `response_schema`; `json_schema` sends its raw JSON Schema. |
| `_RETRY_CAPACITY_ERRORS_INDEFINITELY` | `false`            | If `true`, retry capacity (overload) errors without a retry limit. |

Example environment variables:

```env
GEMINI_API_KEYS=key1,key2,key3
GEMINI_MODEL=gemini-3.5-flash-lite
GEMINI_FALLBACK_MODELS=gemini-3.1-flash-lite
GEMINI_PROXY_URL=http://proxy.example:18888
GEMINI_RPM=15
GEMINI_TPM=250000
GEMINI_RPD=500
GEMINI_MAX_OUTPUT_TOKENS=2048
```

> `GeminiGatewayConfig.from_env()` reads existing environment variables by
> default. If you want it to load a local `.env` file first, install
> `gemini-gateway[dotenv]` and pass `load_dotenv_file=True`.

### Per-project prefixes

Because the gateway is meant to be shared, each project can isolate its
configuration with a custom prefix:

```python
from gemini_gateway import GeminiGateway, GeminiGatewayConfig

config = GeminiGatewayConfig.from_env(prefix="PUBLISHER_GEMINI")
gateway = GeminiGateway(config)
```

This reads `PUBLISHER_GEMINI_API_KEYS`, `PUBLISHER_GEMINI_MODEL`, and so on.

### Configuring in code

You can skip the environment entirely and build a config directly. Invalid
values are rejected at construction time with a clear `ValueError`.

```python
from gemini_gateway import GeminiGateway, GeminiGatewayConfig

config = GeminiGatewayConfig(
    model="gemini-3.1-flash-lite",
    api_keys=("key1", "key2"),
    requests_per_minute=10,
    tokens_per_minute=200_000,
    requests_per_day=400,
)
gateway = GeminiGateway(config)
```

## Model fallback

`fallback_models` turns extra models into overflow capacity for the primary one:

```python
config = GeminiGatewayConfig(
    model="gemini-3.5-flash-lite",
    fallback_models=("gemini-3.1-flash-lite",),
    api_keys=("key1", "key2"),
)
```

Models are tried in order, and a model is only skipped when **no** key can serve
it right now. So the gateway exhausts `gemini-3.5-flash-lite` on every key before
sending anything to `gemini-3.1-flash-lite`. `GeminiTextResult.model` /
`GeminiJsonResult.model` report which model actually answered, and the key label
becomes `key-1/gemini-3.5-flash-lite` so logs stay unambiguous.

## How rate limiting works

Limits are enforced per **`(key, model)` pair** by `MultiKeyRateLimiter` before a
request is sent. Google grants free-tier quota per Cloud project *and* per model,
so the pair is the smallest independent quota bucket: exhausting the daily quota
of one model on one key leaves every other key — and every other model of the
same key — untouched.

On each call the limiter walks models in configured order and keys in round-robin
order, skipping any slot that is disabled, cooling down, or would exceed its RPM,
TPM, or RPD window. If every slot is momentarily unavailable, the limiter sleeps
until the soonest one frees up rather than overshooting quota.

Cooldown scope follows the error:

| Error                                      | Scope                                    | Duration                              |
| ------------------------------------------ | ---------------------------------------- | ------------------------------------- |
| `…PerDayPerProjectPerModel` quota           | the failing `(key, model)` pair           | `daily_quota_cooldown_seconds` (hint is a lower bound) |
| RPM/TPM quota                               | the failing `(key, model)` pair           | provider `retryDelay`, else 60 s      |
| `5xx` / network                             | the failing `(key, model)` pair           | provider `retryDelay`, else 5 s       |
| `401` / `403`                               | the whole key, all its models             | permanent (`disable_key`)             |

`availability()` returns a secret-free `KeyAvailability` snapshot — slot counts,
the nearest `next_available_at`, and per-slot block reasons using labels only, so
it is safe to log:

```python
availability = gateway.rate_limiter.availability()
if not availability.has_available_key:
    logger.warning("waiting %.0fs: %s", availability.wait_seconds(), availability.describe())
```

Token accounting uses an estimate of `prompt_length / 4 + max_output_tokens`
when you don't pass an explicit `token_budget`. RPD is tracked as a rolling
24-hour window — an API-safe approximation, not a calendar-day reset at the
provider's midnight.

## Driving your own retry loop

Callers that own batching, persistence or splitting logic can use the limiter
directly and skip the built-in retry loop:

```python
state = gateway.rate_limiter.acquire(token_budget=12_000)
try:
    result = gateway.generate_json_on_slot(state, prompt, BatchOut, max_output_tokens=4096)
except Exception as exc:
    error_info = classify_api_error(exc)
    if error_info.should_cooldown_key:
        gateway.rate_limiter.cooldown_key_after_error(state.key, error_info, model=state.model)
    raise
```

## Retries and error handling

Each failed attempt is classified by `classify_api_error`:

| Category         | Retryable | Notes                                                          |
| ---------------- | --------- | -------------------------------------------------------------- |
| `rate_limit`     | yes       | Cools down the `(key, model)` pair; honors `retryDelay` hints. `quota_scope` / `quota_id` say whether the daily project quota was hit. |
| `network`        | yes       | Connection/timeout/TLS errors.                                 |
| `server`         | yes       | `5xx` / overload. Capacity errors are retryable indefinitely only when opted in. |
| `empty_response` | yes       | Model returned no text; retried on the next slot.              |
| `content_policy` | mixed     | `RECITATION` is retried, `PROHIBITED_CONTENT` is not; both hint `should_split`. |
| `auth`           | no        | Disables the key (all its models) and rotates to the next one. |
| `validation`     | mixed     | Malformed model output is retried with `should_split`; bad configuration is not. |
| `internal`       | no        | Programming errors (`TypeError`, `KeyError`, …).               |
| `no_active_keys` | no        | Every key is disabled.                                         |
| `unknown`        | yes       | Conservatively retried.                                        |

`ApiErrorInfo.should_split` is advisory: it tells a batching caller that retrying
a smaller request is likely to help.

Retries use exponential backoff (`retry_base_delay_seconds * 2**attempt`, capped
at `retry_max_delay_seconds`) and respect any server-provided delay hint. When
all retries are exhausted, a `GeminiRetriesExhaustedError` is raised with the
last underlying exception chained.

```python
from gemini_gateway import GeminiGateway, GeminiRetriesExhaustedError, NoApiKeysError

try:
    gateway = GeminiGateway.from_env()
    text = gateway.generate_text("Hello!")
except NoApiKeysError:
    ...  # no keys configured
except GeminiRetriesExhaustedError as exc:
    ...  # transient failures persisted past max_retries; exc.__cause__ has details
```

## Testing your own code

The network client, clock, and sleep function are injectable, so you can drive
the gateway deterministically without any network access:

```python
from types import SimpleNamespace
from gemini_gateway import GeminiGateway, GeminiGatewayConfig


class FakeModels:
    def generate_content(self, **kwargs):
        return SimpleNamespace(text="stubbed", usage_metadata=None)


class FakeClient:
    models = FakeModels()


config = GeminiGatewayConfig(model="gemini-test", api_keys=("k",))
gateway = GeminiGateway(
    config,
    client_factory=lambda api_key, timeout_ms: FakeClient(),
    sleep_fn=lambda _seconds: None,  # no real waiting
)

assert gateway.generate_text("hi") == "stubbed"
```

## Public API

Everything below is exported from the top-level `gemini_gateway` package.

**Entry points**

- `GeminiGateway` — `from_env(prefix="GEMINI")`, `generate_text`,
  `generate_text_result`, `generate_json`, `generate_json_result`,
  `generate_text_on_slot`, `generate_json_on_slot`, `count_tokens`,
  `rate_limiter`
- `GeminiGatewayConfig` — `from_env(prefix=..., env=..., load_dotenv_file=...)`,
  `models`

**Result types**

- `GeminiTextResult`, `GeminiJsonResult`, `GeminiUsage`

**Rate limiting**

- `MultiKeyRateLimiter` — `acquire`, `has_ready_key`, `availability`,
  `cooldown_key`, `cooldown_key_after_error`, `disable_key`,
  `restore_disabled_keys`
- `KeyState`, `KeyAvailability`

**Errors & classification**

- `classify_api_error`, `ApiErrorInfo`
- `QUOTA_SCOPE_PER_KEY_DAILY`, `QUOTA_SCOPE_PER_KEY_WINDOW`
- `GeminiGatewayError` (base), `NoApiKeysError`, `AllApiKeysDisabledError`,
  `GeminiRetriesExhaustedError`

The package ships a `py.typed` marker, so type checkers see the annotations.

## Development

```bash
python -m pip install -e ".[dev]"
python -m ruff check .
python -m ruff format --check .
python -m pyright
python -m pytest
python -m build
python -m twine check dist/*
```

Type checking is done by Pyright in `strict` mode for `src` (configured in
`[tool.pyright]`); `tests` are excluded to avoid mock-related false positives.
Suppressions (`# pyright: ignore[...]`) must be narrow and explain why the code
is valid.

Tests use fake clients and never call the real Gemini API. Contributions should
keep the public API small and typed, and avoid adding application-specific logic.

## License

Apache-2.0. See [LICENSE](LICENSE).
