Metadata-Version: 2.4
Name: gemini-gateway
Version: 0.1.0
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
Requires-Dist: python-dotenv>=1.0.1
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: mypy>=1.13.0; 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.svg)](https://pypi.org/project/gemini-gateway/)
[![Python](https://img.shields.io/pypi/pyversions/gemini-gateway.svg)](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-2.0-flash"
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-2.0-flash` | Model name.                                                        |
| `_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`                         | `120000`           | 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.2`              | Sampling temperature (0–2).                                        |
| `_RETRY_BASE_SECONDS`                 | `5.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.        |
| `_RETRY_CAPACITY_ERRORS_INDEFINITELY` | `false`            | If `true`, retry capacity (overload) errors without a retry limit. |

Example `.env` (see [`.env.example`](.env.example)):

```env
GEMINI_API_KEYS=key1,key2,key3
GEMINI_MODEL=gemini-2.0-flash
GEMINI_RPM=15
GEMINI_TPM=250000
GEMINI_RPD=500
GEMINI_MAX_OUTPUT_TOKENS=2048
```

> `GeminiGatewayConfig.from_env()` calls `load_dotenv()` by default, which reads
> a `.env` file into `os.environ`. Pass `load_dotenv_file=False` if you need
> strict control over the environment.

### 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-2.0-flash",
    api_keys=("key1", "key2"),
    requests_per_minute=10,
    tokens_per_minute=200_000,
    requests_per_day=400,
)
gateway = GeminiGateway(config)
```

## How rate limiting works

Limits are enforced **per key** by `MultiKeyRateLimiter` before a request is
sent. On each call the limiter picks the next eligible key in round-robin order,
skipping any key that is disabled, cooling down, or would exceed its RPM, TPM, or
RPD window. If every key is momentarily unavailable, the limiter sleeps until the
soonest one frees up rather than overshooting quota.

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.

## Retries and error handling

Each failed attempt is classified by `classify_api_error`:

| Category     | Retryable | Notes                                                          |
| ------------ | --------- | -------------------------------------------------------------- |
| `rate_limit` | yes       | Triggers a key cooldown; honors `retryDelay` hints.            |
| `network`    | yes       | Connection/timeout/TLS errors.                                 |
| `server`     | yes       | `5xx` / overload. Capacity errors are retryable indefinitely only when opted in. |
| `auth`       | no        | Disables the key and rotates to the next available one.        |
| `validation` | no        | Bad/empty/invalid output; raised to the caller.               |
| `internal`   | no        | Programming errors (`TypeError`, `KeyError`, …).               |
| `unknown`    | yes       | Conservatively retried.                                        |

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`
- `GeminiGatewayConfig` — `from_env(prefix=..., env=..., load_dotenv_file=...)`

**Result types**

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

**Rate limiting**

- `MultiKeyRateLimiter`, `KeyState`

**Errors & classification**

- `classify_api_error`, `ApiErrorInfo`
- `GeminiGatewayError` (base), `NoApiKeysError`, `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 mypy src
python -m pytest
python -m build
python -m twine check dist/*
```

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).
