# Feeds

Stream each feed's real-time events and download its hourly and daily Parquet exports.

Each feed is an attribute of the client. On `Client` it is a `Feed`, and on `AsyncClient` an `AsyncFeed`, generic in the feed's event type, such as `Feed[ProxyEvent]`. Both classes are exported for type annotations, but don't create feeds yourself.

Every feed has a real-time [stream](#streams) and hourly Parquet [exports](#exports) that roll up into daily exports. Streaming and exports need separate scopes on the API key. The event types are listed under [Event types](https://docs.synthient.com/sdks/python/types#event-types).

| Attribute      | Event type         | Stream scope            | Export scope          |
| -------------- | ------------------ | ----------------------- | --------------------- |
| `proxies`      | `ProxyEvent`       | `PROXY_FIREHOSE`        | `PROXY_FEEDS`         |
| `anonymizers`  | `AnonymizerEvent`  | `ANONYMIZERS_STREAM`    | `ANONYMIZERS_FEED`    |
| `torrents`     | `TorrentEvent`     | `TORRENTS_STREAM`       | `TORRENTS_FEED`       |
| `ja4t`         | `JA4TEvent`        | `JA4T_STREAM`           | `JA4T_FEED`           |
| `helios_http`  | `HeliosHTTPEvent`  | `HONEYPOT_HTTP_STREAM`  | `HONEYPOT_HTTP_FEED`  |
| `helios_https` | `HeliosHTTPSEvent` | `HONEYPOT_HTTPS_STREAM` | `HONEYPOT_HTTPS_FEED` |
| `helios_dns`   | `HeliosDNSEvent`   | `HONEYPOT_DNS_STREAM`   | `HONEYPOT_DNS_FEED`   |
| `helios_adb`   | `HeliosADBEvent`   | `HONEYPOT_ADB_STREAM`   | `HONEYPOT_ADB_FEED`   |

What each feed carries:

- `proxies` carries residential, mobile, and datacenter proxy sightings.
- `anonymizers` carries VPN and anonymizer IP ranges.
- `torrents` carries torrents seen on the BitTorrent DHT and trackers.
- `ja4t` carries JA4T TCP fingerprints.
- The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, tunnel destinations, and Android Debug Bridge commands. Despite its name, `helios_dns` does not carry DNS queries.

See [Firehose](https://docs.synthient.com/enterprise/firehose), [Feeds](https://docs.synthient.com/enterprise/feeds), and [Helios](https://docs.synthient.com/enterprise/helios) for the data behind each feed.

## Streams

### stream

Opens the feed's real-time stream and yields an iterator over its events, where `ModelT` is the feed's event type.

```python
@contextmanager
def stream(self) -> Generator[Iterator[ModelT], None, None]

@asynccontextmanager
async def stream(self) -> AsyncGenerator[AsyncIterator[ModelT], None]
```

- **Connection:** The connection opens when the `with` block is entered and closes when it exits, even after a `break`. The server keeps a stream open for up to 30 minutes and then closes it, which ends the iteration without an error. Opening a stream is retried like any other feed request.
- **Errors:** If the stream cannot be opened, entering the block raises an `APIError` for a non-success status, such as 403 without the feed's scope or 429 when the key already has too many open streams, and `APIConnectionError` or `APITimeoutError` when the connection fails. A connection that fails during iteration raises `APIConnectionError`, and an event that cannot be decoded raises `SynthientError`.
- **Reconnecting:** The API has no resume cursor, so events sent between connections are missed.

Open the stream again to reconnect:

```python
while True:
    with client.proxies.stream() as events:
        for event in events:
            print(event.ip, event.provider, event.type)
```

The loop reconnects immediately after a clean close. Opening a connection already uses the SDK's bounded retries, including `Retry-After` handling. Any error that remains after those retries, or occurs while reading the stream, stops this consumer. A `429` without `Retry-After` can indicate an exhausted monthly request budget or too many open streams; resolve that condition before reconnecting.

With `AsyncClient`, enter the stream with `async with` and iterate with `async for`:

```python
async with client.proxies.stream() as events:
    async for event in events:
        print(event.timestamp, event.ip, event.provider, event.type)
```

Every feed streams the same way. Only the event type changes:

**Streams**

```python title="Proxies"
with client.proxies.stream() as events:
    for event in events:
        print(event.ip, event.provider, event.type, event.country_code, event.asn)
```

```python title="Anonymizers"
with client.anonymizers.stream() as events:
    for event in events:
        print(event.range_start, event.range_end, event.provider, event.type)
```

```python title="Torrents"
with client.torrents.stream() as events:
    for event in events:
        print(event.info_hash, event.name, event.total_size)
```

```python title="JA4T"
with client.ja4t.stream() as events:
    for event in events:
        print(event.ip, event.ja4t, event.provider)
```

```python title="Helios HTTP"
with client.helios_http.stream() as events:
    for event in events:
        if event.details is None:
            continue
        print(event.domain, event.details.method, event.details.uri, event.meta.provider)
```

```python title="Helios HTTPS"
with client.helios_https.stream() as events:
    for event in events:
        if event.details is None:
            continue
        print(event.domain, event.details.sni, event.details.handshake_version)
```

```python title="Helios DNS"
with client.helios_dns.stream() as events:
    for event in events:
        print(event.domain, event.port, event.meta.provider)
```

```python title="Helios ADB"
with client.helios_adb.stream() as events:
    for event in events:
        print(event.session, event.sequential_id, event.command)
```

`HeliosHTTPEvent.details` and `HeliosHTTPSEvent.details` are `None` when there is no parsed request or handshake, so check them before use.

## Exports

Each feed has hourly Parquet exports that roll up into daily exports.

- **Hourly exports:** An hourly export is available until its day's daily export is rolled up, shortly after 00:30 UTC the next day. After that, requests for the hour raise an `APIError` with status 400.
- **Daily exports:** While a daily export is being rolled up, requests for it raise an `APIError` with status 404 and `retry_after` set.

### Export IDs

Methods that take an `export` accept:

- `"latest"`, the feed's freshest export: the newest hourly export during the day, or the daily export once that day's hours are rolled up.
- A UTC day as `"YYYY-MM-DD"` or a `date`, for a daily export.
- A UTC hour as `"YYYY-MM-DD/H"`, where a zero-padded hour is also accepted, or a timezone-aware `datetime`, for an hourly export. The `datetime` is converted to UTC.

Notes on these IDs:

- **Input:** `Export.id` values can be passed as they are. There are no named helpers for these IDs.
- **Datetimes:** Because `datetime` is a subclass of `date`, a `datetime` always selects an hourly export, so call `.astimezone(UTC).date()` on it to select its UTC day.
- **Errors:** A naive `datetime` or any other string raises `ValueError`, and an export ID of another type raises `TypeError`, before a request is sent.

```python
from datetime import UTC, datetime, timedelta

now = datetime.now(UTC)
yesterday = client.proxies.export_meta((now - timedelta(days=1)).date())
last_hour = client.proxies.export_meta(now - timedelta(hours=1))
print(yesterday.id, last_hour.id)
```

### list\_exports

Returns one page of the feed's exports, newest first.

```python
def list_exports(self, *, limit: int | None = None, cursor: str | None = None) -> ExportPage

async def list_exports(
    self, *, limit: int | None = None, cursor: str | None = None
) -> ExportPage
```

- `limit` `int | None` — The page size, from 1 to 500. `None` uses the API default of 100. A value below 1, or one that is not an integer, raises `ValueError`.
- `cursor` `str | None` — Continues a listing from a previous page's `next_cursor`.

```python
page = client.proxies.list_exports(limit=24)
for export in page.exports:
    print(export.id, export.kind, export.row_count)
print("next cursor:", page.next_cursor)
```

- **Paging:** Use `iter_exports` to iterate over every page.

### iter\_exports

Iterates over the feed's exports, newest first.

```python
def iter_exports(
    self, *, limit: int | None = None, cursor: str | None = None
) -> Iterator[Export]

async def iter_exports(
    self, *, limit: int | None = None, cursor: str | None = None
) -> AsyncIterator[Export]
```

```python
for export in client.proxies.iter_exports():
    print(export.id, export.kind, export.size_bytes)
```

On `AsyncFeed`, call it without `await` and iterate with `async for`:

```python
async for export in client.proxies.iter_exports():
    print(export.id, export.kind, export.size_bytes)
```

- **Paging:** It fetches pages of `limit` lazily as iteration proceeds, starting from `cursor`. It stops fetching when the loop breaks.
- **Results:** On the first page, the newest hourly export has the ID `"latest"`, a moving alias for the freshest export. To keep a stable ID, build it from the export's `date` and `hour`, such as `"2026-10-14/21"`.
- **Errors:** If the API repeats a cursor, it raises `SynthientError` after that page's exports.

### export\_meta

Returns the checksum, size, row count, and Parquet schema of an export.

```python
def export_meta(self, export: str | date) -> ExportMeta

async def export_meta(self, export: str | date) -> ExportMeta
```

```python
meta = client.proxies.export_meta("latest")
print(meta.id, meta.rows, meta.size, meta.checksum)
for column in meta.parquet_schema.fields:
    print(column.name, column.type)
```

### download\_url

Returns a presigned URL for an export's Parquet file.

```python
def download_url(self, export: str | date) -> str

async def download_url(self, export: str | date) -> str
```

```python
from urllib.parse import urlsplit

url = client.proxies.download_url("latest")
print(urlsplit(url).hostname)
```

- **Lifetime:** The URL is valid for 24 hours.
- **Security:** The URL grants access without an API key, so treat it as a secret.
- **Use:** It suits tools that read Parquet over HTTPS directly.
- **Errors:** A redirect to anything other than an `http` or `https` URL, or from `https` to `http`, raises `SynthientError`.

### download

Downloads an export's Parquet file and yields the streaming `httpx.Response`.

```python
@contextmanager
def download(self, export: str | date) -> Generator[httpx.Response, None, None]

@asynccontextmanager
async def download(self, export: str | date) -> AsyncGenerator[httpx.Response, None]
```

- **Reading:** Read the body with `response.iter_bytes()`, or `response.aiter_bytes()` on `AsyncFeed`.
- **Cleanup:** The response closes when the block exits.
- **Security:** The API key is sent only to the Synthient API, never to the storage host that serves the file, and errors never include the presigned URL.
- **Errors:** A non-success response from either raises an `APIError`, and a missing or invalid redirect raises `SynthientError`. Reading the body is plain httpx, so a failure partway through raises an httpx exception, such as `httpx.ReadTimeout` or `httpx.ReadError`, rather than a `SynthientError`, and leaves a partial file. Downloads from the storage host are not retried.

The SDK does not verify the file. Compare its SHA-256 with `Export.checksum` or `ExportMeta.checksum`, and ask for the checksum and the file by the same stable ID, because `"latest"` can move to a newer export between the two calls:

```python
import hashlib
from datetime import UTC, date, datetime, timedelta
from pathlib import Path

import synthient


def save_export(feed: synthient.Feed[synthient.ProxyEvent], export: date, path: Path) -> None:
    meta = feed.export_meta(export)
    digest = hashlib.sha256()
    with feed.download(export) as response, path.open("wb") as file:
        for chunk in response.iter_bytes():
            file.write(chunk)
            digest.update(chunk)
    if digest.hexdigest() != meta.checksum:
        raise ValueError("checksum mismatch")


yesterday = datetime.now(UTC).date() - timedelta(days=1)
save_export(client.proxies, yesterday, Path("proxies.parquet"))
```