Metadata-Version: 2.4
Name: datacurrent
Version: 1.0.0
Summary: Python streaming, made simple: async streams, operators, backpressure and real-time transports.
Author: DataCurrent contributors
License: MIT
Keywords: streaming,async,asyncio,real-time,data pipeline,stream processing,backpressure,reactive,event bus,pubsub,websocket,server-sent events,http streaming,concurrency,observability,http,https
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Requires-Dist: websockets>=12
Provides-Extra: starlette
Requires-Dist: starlette>=0.36; extra == "starlette"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == "fastapi"
Provides-Extra: all
Requires-Dist: starlette>=0.36; extra == "all"
Requires-Dist: fastapi>=0.110; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-cov>=5; extra == "dev"
Requires-Dist: hypothesis>=6; extra == "dev"
Requires-Dist: build>=1.2; extra == "dev"
Requires-Dist: twine>=5; extra == "dev"
Requires-Dist: ruff>=0.6; extra == "dev"
Requires-Dist: mypy>=1.11; extra == "dev"
Requires-Dist: fastapi>=0.110; extra == "dev"
Dynamic: license-file

# DataCurrent

### Python streaming, made simple.

DataCurrent is a production-ready asynchronous streaming library. It gives you
one consistent API for lists, async generators, queues, WebSockets, Server-Sent
Events and streaming HTTP responses.

**Version 1.0.0 · Stable / Production**

- 279 tests passing, with 100% line and branch coverage.
- Required networking dependencies install automatically.
- No broker, database, API key or hosted service.

## Install

```bash
pip install datacurrent
```

Python 3.11 or newer is required. `httpx` and `websockets` are installed with the
package, so HTTP streaming, SSE and WebSocket clients work immediately.

Framework helpers are optional:

```bash
pip install "datacurrent[fastapi]"
pip install "datacurrent[starlette]"
pip install "datacurrent[all]"
```

## Quick start

```python
import asyncio
from datacurrent import Stream


async def main() -> None:
    result = await (
        Stream.from_iterable(range(10))
        .filter(lambda number: number % 2 == 0)
        .map(lambda number: number * number)
        .take(4)
        .collect()
    )
    print(result)  # [0, 4, 16, 36]


asyncio.run(main())
```

Streams are lazy and single-consumer:

```python
stream = Stream.from_iterable([1, 2, 3])

async for item in stream:
    print(item)

# A second iteration raises StreamConsumedError.
```

Common terminals:

```python
await Stream.from_iterable([1, 2, 3]).collect()  # [1, 2, 3]
await Stream.from_iterable([1, 2, 3]).count()  # 3
await Stream.from_iterable([3, 1, 2]).reduce(max)  # 3
await Stream.empty().first("fallback")  # "fallback"
```

Async callbacks work everywhere a callback is accepted:

```python
users = await Stream.from_iterable(user_ids).map(fetch_user, concurrency=8).collect()
```

## Operators

`map`, `filter`, `flat_map`, `take`, `skip`, `batch`, `buffer`, `distinct`,
`enumerate`, `window`, `debounce`, `throttle`, `timeout`, `retry`, `recover`,
`tap`, `zip`, `merge` and `concat`.

```python
batches = Stream.from_iterable(range(1_000)).filter(lambda number: number % 3 == 0).batch(100)

async for batch in batches:
    print(len(batch))
```

## Backpressure

Buffers are bounded and block the producer by default. Data is not silently
dropped.

```python
stream = Stream.from_iterable(range(100_000)).buffer(size=64)
```

Lossy behaviour must be explicit and is counted:

```python
stream = Stream.from_iterable(range(10_000)).buffer(
    size=16,
    overflow="drop_oldest",
)
items = await stream.collect()
print(len(items), stream.stats["dropped"])
```

Policies: `block`, `drop_newest`, `drop_oldest`, `error`.

## Real-time data

### WebSocket

```python
from datacurrent import WebSocketStream

async with WebSocketStream.connect("wss://example.com/events") as stream:
    await stream.send_json({"subscribe": "trades"})
    async for message in stream:
        print(message)
```

### Server-Sent Events

```python
from datacurrent import SSEStream

async with SSEStream.connect("https://example.com/events") as stream:
    async for event in stream:
        print(event.event, event.json())
```

### Streaming HTTP

```python
from datacurrent import HTTPStream

async with HTTPStream("https://example.com/feed.ndjson") as response:
    async for line in response.iter_lines():
        handle(line)
```

### Fan-out and events

```python
from datacurrent import Channel, EventBus

channel = Channel(buffer_size=64)
first = channel.subscribe()
second = channel.subscribe()
await channel.publish("deploy finished")

bus = EventBus()


@bus.subscribe("user.*")
async def on_user(event):
    print(event.topic, event.data)


await bus.publish("user.created", {"user_id": 123})
```

## Retries

```python
from datacurrent import RetryPolicy

stream = Stream.from_factory(connect_and_read).retry(
    RetryPolicy(max_attempts=3, base_delay=0.2, jitter=0.1)
)
```

Retries require a replayable source, such as a factory or a list. A consumed
generator raises `NonReplayableStreamError` instead of silently restarting.

## Why DataCurrent

- One model for iterables, queues, HTTP, SSE and WebSockets.
- Backpressure by default, with explicit and counted drop policies.
- Bounded concurrency and cancellation-safe cleanup.
- No infrastructure: no broker, database or external account.
- Explicit delivery semantics; no false durability claims.
- Typed API and sync or async callbacks throughout.

HTTP clients such as `httpx` and `aiohttp` handle requests; DataCurrent adds the
stream engine around them. Reactive libraries provide operator algebra, while
broker frameworks require Kafka, Redis or RabbitMQ. DataCurrent is the
in-process streaming layer that works without any of them.

## Compatibility and limits

- Python 3.11+ on asyncio; no Trio or AnyIO backend.
- In-memory `Channel` and `EventBus` delivery is at most once.
- No durable broker adapters or OpenTelemetry exporter in 1.0.0.
- Not published to PyPI; the project name is provisional.

## License

MIT.
