Metadata-Version: 2.4
Name: nexo-client
Version: 5.0.2
Summary: High-performance Python Client for Nexo Broker
Author: Emanuele Pifani
License-Expression: MIT
Project-URL: Repository, https://github.com/emanuel-epifani/nexo.git
Keywords: microservices,event-driven,message-broker,queue,job-queue,stream,event-streaming,event-sourcing,pubsub,realtime,store,key-value-store,python,high-performance,asyncio
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Networking
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# Nexo Client SDK

High-performance Python client for [Nexo](https://nexo-docs-hub.vercel.app/).



## Quick Start


### Run server
```bash
docker run -p 7654:7654 emanuelepifani/nexo:latest
```
This exposes:
- Port 7654 (TCP): Main server socket for SDK clients.

### Install SDK

```bash
uv add nexo-client
```

Or with pip:
```bash
pip install nexo-client
```


### Connection
```python
from nexo import NexoClient

client = await NexoClient.connect(host="localhost", port=7654)
```

### 1. STORE

```python
# Set key
await client.store.map.set("user:1", {"name": "Max", "role": "admin"})
# Get key
user: User | None = await client.store.map.get("user:1")
# Delete key
await client.store.map.delete("user:1")
```

### 2. QUEUE

```python
# Create queue
mail_q: NexoQueue[Email] = await client.queue("emails").create()
# Push message
await mail_q.push({"to": "test@test.com"})
# Subscribe
async def handle_email(msg: Email) -> None:
    print(msg)

await mail_q.subscribe(handle_email)
# Delete queue
await mail_q.delete()
```

### 3. PUB/SUB

```python
# Define topic (no need to create, auto-created on first publish)
alerts: NexoTopic[Alert] = client.pubsub("system-alerts")
# Subscribe
async def on_alert(msg: Alert) -> None:
    print(msg)

await alerts.subscribe(on_alert)
# Publish
await alerts.publish({"level": "high"})
```

### 4. STREAM

```python
# Create topic
stream: NexoStream[UserEvent] = await client.stream("user-events").create()
# Publisher
await stream.publish({"type": "login", "userId": "u1"})
# Consumer (must specify group)
async def on_event(msg: UserEvent, meta: StreamMessageMeta) -> None:
    print(f"User {msg['userId']} performed {msg['type']}")

await stream.subscribe("analytics", on_event)
# Delete topic
await stream.delete()
```

> Callbacks for Queue, Pub/Sub, and Stream can be sync `def` or async `async def` — the SDK handles both.



---

### Binary Payloads

All Nexo brokers (**Store, Queue, Stream, PubSub**) natively support raw binary data (`bytes`).    
Bypassing JSON serialization drastically reduces Latency, increases Throughput, and saves Bandwidth.

**Perfect for:** Video chunks, Images, Protobuf/MsgPack, Encrypted blobs.

```python
# Send 1MB raw bytes (30% smaller than JSON/Base64)
heavy_payload = b"\x00" * (1024 * 1024)

# 1. STREAM
stream: NexoStream[bytes] = client.stream("cctv-archive")
await stream.publish(heavy_payload)
# 2. PUBSUB
audio_topic: NexoTopic[bytes] = client.pubsub("live-audio-call")
audio_topic.publish(heavy_payload)
# 3. STORE
await client.store.map.set("user:avatar:1", heavy_payload)
# 4. QUEUE
queue: NexoQueue[bytes] = client.queue("pdf-processing")
await queue.push(heavy_payload)
```

---

## License

MIT


## Links

- **Full Documentation:** [Nexo Docs](https://nexo-docs-hub.vercel.app/)
- **Docker Image:** [emanuelepifani/nexo](https://hub.docker.com/r/emanuelepifani/nexo)
- **PyPI:** [nexo-client](https://pypi.org/project/nexo-client/)

## Author

Built by **Emanuel Epifani**.

- [LinkedIn](https://www.linkedin.com/in/emanuel-epifani/)
- [GitHub](https://github.com/emanuel-epifani)
