ZnSocket

https://github.com/zincware/znsocket
Admin
ZnSocket is a Redis-compatible key-value store implemented in Python using socket.io, providing...
Tokens:22,984
Snippets:147
Trust Score:7.4
License:Apache-2.0
Update:9 months ago
Show doc for...
Tokens:
Raw
### Set up znsocket Server

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Starts a znsocket server on a specified port. This is the foundational step for any znsocket application, allowing clients to connect and communicate.

```python
import znsocket

# Start the server
server = znsocket.Server(port=5000)
server.run()
```

--------------------------------

### DictAdapter Example (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Demonstrates exposing a Python dictionary as a znsocket Dict adapter, enabling real-time access and modification of the dictionary's contents from other clients.

```python
import znsocket

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Create configuration data
app_config = {
    "database_url": "postgresql://localhost/myapp",
    "debug": False,
    "max_users": 1000
}

# Expose through adapter
adapter = znsocket.DictAdapter(
    socket=client,
    key="app_settings",
    object=app_config
)

# Access from any client
settings = znsocket.Dict(client, "app_settings")
print(f"Database: {settings['database_url']}")
print(f"Debug mode: {settings['debug']}")

# Real-time configuration updates
app_config["debug"] = True
print(f"Debug now: {settings['debug']}")  # Immediately updated
```

--------------------------------

### Real-time Synchronization Example (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Illustrates how changes made to a shared data structure on one client are immediately reflected on other connected clients, demonstrating real-time synchronization.

```python
# Client 1
client1 = znsocket.Client("http://localhost:5000")
shared_list = znsocket.List(client1, "shared_data")

# Client 2
client2 = znsocket.Client("http://localhost:5000")
shared_list2 = znsocket.List(client2, "shared_data")

# Changes from client1 are immediately visible to client2
shared_list.append("new_item")
print(shared_list2[-1])  # "new_item"
```

--------------------------------

### Start znsocket Server (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Initializes and runs a znsocket server on a specified port. This server will handle client connections and data synchronization.

```python
import znsocket

# Start server on port 5000
server = znsocket.Server(port=5000)
server.run()
```

--------------------------------

### Start znsocket server

Source: https://github.com/zincware/znsocket/blob/main/README.md

Starts the znsocket server from the command line, specifying the port to listen on. Use --help for additional configuration options.

```bash
znsocket --port 5000
```

--------------------------------

### Start ZnSocket Server (Bash)

Source: https://context7.com/zincware/znsocket/llms.txt

Starts the ZnSocket server with various configurations including default settings, custom ports, buffer sizes, Redis backend, and debug logging.

```bash
# Start server with default settings (port 8080, memory storage)
znsocket

# Start server on custom port with larger buffer size
znsocket --port 5000 --max-http-buffer-size 10485760

# Start server with Redis backend
znsocket --port 5000 --storage redis://localhost:6379

# Start server with debug logging
znsocket --port 5000 --log-level DEBUG
```

--------------------------------

### ListAdapter Example (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Exposes a Python list-like object as a znsocket List adapter, allowing it to be accessed and modified in real-time by other clients without data copying.

```python
import znsocket
import numpy as np

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Create some data (any list-like object)
sensor_data = [23.5, 24.1, 23.8, 24.0, 23.9]
# Or: sensor_data = np.array([23.5, 24.1, 23.8, 24.0, 23.9])

# Expose through adapter
adapter = znsocket.ListAdapter(
    socket=client,
    key="temperature_readings",
    object=sensor_data
)

# Access from any client
readings = znsocket.List(client, "temperature_readings")
print(f"Current temperature: {readings[-1]}°C")
print(f"Average: {sum(readings) / len(readings):.1f}°C")

# Real-time updates: modify original data
sensor_data.append(24.2)
print(f"New reading: {readings[-1]}°C")  # Immediately available
```

--------------------------------

### Install znsocket with npm (Bash)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Installs the znsocket library for JavaScript projects using the npm package manager. Ensure Node.js and npm are installed and accessible.

```bash
npm install znsocket
```

--------------------------------

### Python: Dynamic DictAdapter for System Status

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This example shows how to use DictAdapter to expose dynamic system status information. It creates a dictionary for system metrics and updates it in a background thread, making real-time status available via the 'system_status' key.

```python
import znsocket
import time
import threading
import random

# Create dynamic configuration
system_status = {
    "cpu_usage": 0.0,
    "memory_usage": 0.0,
    "disk_usage": 0.0,
    "active_connections": 0
}

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose status through adapter
adapter = znsocket.DictAdapter(
    key="system_status",
    socket=client,
    object=system_status
)

# Simulate system monitoring
def update_system_status():
    while True:
        system_status["cpu_usage"] = random.uniform(0, 100)
        system_status["memory_usage"] = random.uniform(0, 100)
        system_status["disk_usage"] = random.uniform(0, 100)
        system_status["active_connections"] = random.randint(0, 500)
        time.sleep(1)  # Update every second

# Start monitoring in background
thread = threading.Thread(target=update_system_status)
thread.daemon = True
thread.start()

print("System status available at key: system_status")
```

--------------------------------

### Python: Consume Data with List Interface and Slicing

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This example demonstrates how to consume data exposed via the 'scientific_data' key using znsocket's List interface. It shows how to get the data length, access individual items, retrieve all items, and perform various slicing operations.

```python
import znsocket

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Access data through List interface
data = znsocket.List(client, "scientific_data")

# Read data with various operations
print(f"Data length: {len(data)}")
print(f"First item: {data[0]}")
print(f"All items: {data[:]}")

# Efficient slicing operations
print(f"First 3 items: {data[:3]}")
print(f"Every 2nd item: {data[::2]}")
print(f"Last 3 items: {data[-3:]}")
print(f"Items 2-4: {data[1:4]}")
```

--------------------------------

### JavaScript: Client Access to List and Dict Data

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This JavaScript example shows how to connect to the znsocket server and access data using the client library. It demonstrates accessing sensor readings via a List interface and system status/app configuration via Dict interfaces.

```javascript
import { createClient, List, Dict } from 'znsocket';

// Connect to server
const client = createClient({ url: 'http://localhost:5000' });
await client.connect();

// Access sensor data
const sensorData = new List({ client, key: 'sensor_readings' });
console.log(`Sensor readings: ${await sensorData.length()}`);
console.log(`Latest 5 readings: ${await sensorData.slice(-5)}`);

// Access system status
const systemStatus = new Dict({ client, key: 'system_status' });
console.log(`CPU Usage: ${await systemStatus.get('cpu_usage')}%`);
console.log(`Memory Usage: ${await systemStatus.get('memory_usage')}%`);

// Access configuration
const appConfig = new Dict({ client, key: 'app_config' });
const dbHost = await appConfig.get('database_host');
console.log(`Database host: ${dbHost}`);
```

--------------------------------

### Install ZnSocket (Bash)

Source: https://context7.com/zincware/znsocket/llms.txt

Installs the ZnSocket library for Python using pip and for JavaScript/Node.js using npm.

```bash
# Python installation
pip install znsocket

# JavaScript/Node.js installation
npm install znsocket
```

--------------------------------

### Python Client: Connect and Use Data Structures (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Demonstrates connecting a Python client to a znsocket server and creating/manipulating distributed list and dictionary data structures.

```python
import znsocket

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Create a distributed list
my_list = znsocket.List(client, "my_list")
my_list.append("item1")
my_list.append("item2")

# Create a distributed dict
my_dict = znsocket.Dict(client, "my_dict")
my_dict["key1"] = "value1"
my_dict["key2"] = "value2"
```

--------------------------------

### Install znsocket with pip (Bash)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Installs the znsocket library for Python projects using the pip package manager. Ensure pip is installed and accessible in your environment.

```bash
pip install znsocket
```

--------------------------------

### Python: Expose Configuration with DictAdapter

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This snippet illustrates how to expose a Python dictionary as configuration data using znsocket's DictAdapter. It connects to the server and creates an adapter for 'app_config' with predefined configuration settings.

```python
import znsocket

# Create configuration data
config = {
    "database_host": "localhost",
    "database_port": 5432,
    "cache_enabled": True,
    "max_connections": 100
}

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose config through adapter
adapter = znsocket.DictAdapter(
    key="app_config",
    socket=client,
    object=config
)

print("Configuration exposed through adapter")
```

--------------------------------

### Nest Dictionaries within a List using znsocket

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Demonstrates creating a znsocket list ('users_list') and appending znsocket dictionaries ('user1', 'user2') to it. This showcases how to manage collections of structured records, where each record is a dictionary within a list.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Create a main list to store user records
users_list = znsocket.List(client, "users_list")

# Create individual user dictionaries
user1 = znsocket.Dict(client, "user1")
user1.update({
    "name": "Alice",
    "age": 30,
    "department": "Engineering"
})

user2 = znsocket.Dict(client, "user2")
user2.update({
    "name": "Bob",
    "age": 25,
    "department": "Design"
})

# Store dictionaries in the list
users_list.append(user1)  # Dict inside List
users_list.append(user2)  # Another Dict inside List

# Access nested data
print(f"First user name: {users_list[0]['name']}")
print(f"Second user department: {users_list[1]['department']}")
```

--------------------------------

### Python: Consume Configuration with Dict Interface

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This snippet demonstrates how to consume configuration data exposed via the 'app_config' key using znsocket's Dict interface. It shows how to access individual configuration values, check for key existence, and retrieve all keys and values.

```python
import znsocket

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Access configuration through Dict interface
config = znsocket.Dict(client, "app_config")

# Read configuration
print(f"Database host: {config['database_host']}")
print(f"Database port: {config['database_port']}")
print(f"Cache enabled: {config['cache_enabled']}")

# Check if keys exist
if "debug_mode" in config:
    print(f"Debug mode: {config['debug_mode']}")

# Get all keys and values
print(f"All keys: {list(config.keys())}")
print(f"All values: {list(config.values())}")
```

--------------------------------

### Python Configuration Server

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

A Python script that acts as a configuration server using znsocket. It initializes a shared dictionary named 'app_config' with initial configuration values.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")
config = znsocket.Dict(client, "app_config")

# Set initial configuration
config.update({
    "database_url": "postgresql://localhost/myapp",
    "cache_enabled": True,
    "max_connections": 100,
    "debug": False
})

print("Configuration initialized")
```

--------------------------------

### Run JS/TS Tests with Docker and Bun

Source: https://github.com/zincware/znsocket/blob/main/docs/source/develop.rst

This snippet explains how to run JavaScript/TypeScript tests for ZNSocket. It involves starting a Docker container with the znsocket image and then using 'bun' to run tests, specifically targeting those with 'native_' in their name.

```bash
docker run --rm -p 4748:4748 pythonf/znsocket
bun test --test-name-pattern native_
```

--------------------------------

### Basic Client Usage in Python

Source: https://github.com/zincware/znsocket/blob/main/README.md

Demonstrates connecting to a znsocket server and performing basic set and get operations on key-value pairs. It uses the Client class to establish a connection and interact with the store.

```python
from znsocket import Client

# Connect to the ZnSocket server
c = Client.from_url("znsocket://127.0.0.1:5000")

# Set and get a value
c.set("name", "Fabian")
assert c.get("name") == "Fabian"
```

--------------------------------

### Python Pipeline for Batch Redis Commands

Source: https://context7.com/zincware/znsocket/llms.txt

Illustrates how to use the Pipeline class in Python to batch multiple Redis commands, reducing network overhead. It shows examples for both hash and list operations, demonstrating how to queue commands and execute them simultaneously. The results of each command are returned in order.

```python
from znsocket import Client

client = Client.from_url("znsocket://127.0.0.1:5000")

# Create pipeline for batch operations
pipe = client.pipeline()

# Queue multiple commands
pipe.hset("user:1", "name", "Alice")
pipe.hset("user:1", "email", "alice@example.com")
pipe.hset("user:2", "name", "Bob")
pipe.hset("user:2", "email", "bob@example.com")
pipe.hget("user:1", "name")
pipe.hget("user:2", "name")

# Execute all commands at once
results = pipe.execute()
# Returns: [1, 1, 1, 1, "Alice", "Bob"]

# Pipelines work with list operations too
pipe = client.pipeline()
pipe.rpush("items", "a")
pipe.rpush("items", "b")
pipe.rpush("items", "c")
pipe.lrange("items", 0, -1)
results = pipe.execute()
# Returns: [1, 2, 3, ["a", "b", "c"]]
```

--------------------------------

### Create and Access Nested Dict/List Structures in Python

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Demonstrates creating a nested data structure with dictionaries and lists using ZNSocket in Python. It shows how to initialize Dict and List objects, populate them with data, and access deeply nested values.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Create a complex nested structure: Dict -> List -> Dict
company = znsocket.Dict(client, "company")
company["name"] = "Tech Corp"
company["founded"] = 2020

# Create departments list
departments = znsocket.List(client, "departments")

# Create individual department dictionaries
engineering = znsocket.Dict(client, "engineering_dept")
engineering.update({
    "name": "Engineering",
    "employees": 50,
    "budget": 2000000
})

marketing = znsocket.Dict(client, "marketing_dept")
marketing.update({
    "name": "Marketing",
    "employees": 20,
    "budget": 500000
})

# Build the nested structure
departments.append(engineering)
departments.append(marketing)
company["departments"] = departments

# Access deeply nested data
print(f"Company: {company['name']}")
print(f"Engineering budget: {company['departments'][0]['budget']}")
print(f"Marketing employees: {company['departments'][1]['employees']}")
```

--------------------------------

### JavaScript Client: Connect and Use Data Structures (JavaScript)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Shows how to connect a JavaScript client to a znsocket server and interact with distributed list and dictionary objects, including asynchronous operations.

```javascript
import { createClient, List, Dict } from 'znsocket';

// Connect to server
const client = createClient({ url: 'http://localhost:5000' });
await client.connect();

// Create a distributed list
const myList = new List({ client, key: 'my_list' });
await myList.push('item1');
await myList.push('item2');

// Create a distributed dict
const myDict = new Dict({ client, key: 'my_dict' });
await myDict.set('key1', 'value1');
await myDict.set('key2', 'value2');
```

--------------------------------

### Connect and Access Nested User Profiles with znsocket

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Demonstrates how to connect to a znsocket server, expose a nested Python dictionary as 'user_profiles', and access/modify its data. It uses DictAdapter to manage the data exposure and Dict to access nested elements.

```python
import znsocket

# Sample nested data structure
user_profiles = {
    "user1": {
        "name": "Alice",
        "scores": [85, 92, 78],
        "preferences": {"theme": "light", "notifications": False}
    },
    "user2": {
        "name": "Bob",
        "scores": [90, 88, 95],
        "preferences": {"theme": "dark", "notifications": True}
    }
}

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose the nested structure
adapter = znsocket.DictAdapter(
    key="user_profiles",
    socket=client,
    object=user_profiles
)

# Access nested data
profiles = znsocket.Dict(client, "user_profiles")

# Direct access to nested structures
user1 = profiles["user1"]
print(f"User 1 name: {user1['name']}")
print(f"User 1 scores: {user1['scores']}")
print(f"User 1 theme: {user1['preferences']['theme']}")

# Modify the original data - changes are immediately visible
user_profiles["user1"]["scores"].append(99)
print(f"Updated scores: {profiles['user1']['scores']}")
```

--------------------------------

### Python ListAdapter and Segments Setup with ZnSocket

Source: https://github.com/zincware/znsocket/blob/main/docs/source/copy_on_write.rst

Demonstrates setting up a Python ListAdapter to expose a Python list via ZnSocket. It then shows how to create a List view and a copy-on-write Segments view, and how to modify data within Segments, showcasing independent data coexistence.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Start with a regular Python list
original_data = [
    {"name": "item_0", "score": 85, "category": "A"},
    {"name": "item_1", "score": 92, "category": "B"},
    {"name": "item_2", "score": 78, "category": "A"},
    {"name": "item_3", "score": 96, "category": "C"},
    {"name": "item_4", "score": 83, "category": "B"},
]

# Use ListAdapter to expose Python list via ZnSocket
znsocket.ListAdapter(
    socket=client,
    key="test:adapter_data",
    object=original_data
)

# Create a List view of the adapted data
lst = znsocket.List(r=client, key="test:adapter_data")

# Create copy-on-write view using Segments
segments = znsocket.Segments(
    r=client,
    origin=lst,
    key="test:adapter_segments"
)

# Create modified versions using copy-on-write
modified_dict = znsocket.Dict(r=client, key="test:adapter_segments/2")
modified_dict.clear()
modified_dict.update({
    "name": "item_2_modified",
    "score": 95,
    "category": "A+",
    "modified": True,
    "source": "segments_copy"
})
segments[2] = modified_dict

# Original Python list remains unchanged: original_data[2]["score"] == 78
# Adapter list remains unchanged: lst[2]["score"] == 78
# Segments shows modification: segments[2]["score"] == 95
```

--------------------------------

### Python ListAdapter for Data Synchronization

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

An example of using znsocket's ListAdapter in Python to expose a NumPy array as a shared resource. This allows real-time access to the array from other clients, including JavaScript.

```python
import znsocket
import numpy as np

# Create some data
data = np.array([1, 2, 3, 4, 5])

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose data through adapter
adapter = znsocket.ListAdapter(
    key="scientific_data",
    socket=client,
    object=data
)

print("Data exposed through adapter")
```

--------------------------------

### Nest Lists within a Dictionary using znsocket

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Illustrates how to create a znsocket dictionary ('user_data') and store znsocket lists ('user_scores', 'user_tags') as values within it. This demonstrates nesting lists inside dictionaries for structured data management.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Create a main dictionary
user_data = znsocket.Dict(client, "user_data")

# Create lists for different user attributes
user_scores = znsocket.List(client, "user_scores")
user_scores.extend([85, 92, 78, 95, 88])

user_tags = znsocket.List(client, "user_tags")
user_tags.extend(["python", "javascript", "redis", "websockets"])

# Store lists in the dictionary
user_data["name"] = "John Doe"
user_data["email"] = "john@example.com"
user_data["scores"] = user_scores  # List inside Dict
user_data["tags"] = user_tags      # Another List inside Dict

# Access nested data
print(f"User name: {user_data['name']}")
print(f"First score: {user_data['scores'][0]}")
print(f"All tags: {user_data['tags'][:]}")
```

--------------------------------

### Create and Access Nested Dict/List Structures in JavaScript

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Illustrates creating a nested data structure with dictionaries and lists using ZNSocket in JavaScript. It covers connecting to the client, creating Dict and List instances, adding data, and retrieving nested information.

```javascript
import { createClient, List, Dict } from 'znsocket';

// Connect to server
const client = createClient({ url: 'http://localhost:5000' });
await client.connect();

// Create nested structure: Dict with List inside
const userProfile = new Dict({ client, key: 'user_profile' });
const userHobbies = new List({ client, key: 'user_hobbies' });

// Add hobbies to the list
await userHobbies.push('coding');
await userHobbies.push('gaming');
await userHobbies.push('reading');

// Store basic info and nested list in dict
await userProfile.set('username', 'developer123');
await userProfile.set('hobbies', userHobbies);

// Access nested data
const username = await userProfile.get('username');
const hobbies = await userProfile.get('hobbies');
const firstHobby = await hobbies.get(0);

console.log(`User: ${username}`);
console.log(`First hobby: ${firstHobby}`);
```

--------------------------------

### Python: Expose Sensor Data with ListAdapter

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This snippet shows how to expose sensor data using znsocket's ListAdapter. It connects to a server, creates an adapter for 'sensor_readings', and simulates real-time data updates in a background thread.

```python
import znsocket
import time
import threading
import random

# Simulate sensor data
sensor_data = []

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose data through adapter
adapter = znsocket.ListAdapter(
    key="sensor_readings",
    socket=client,
    object=sensor_data
)

# Simulate real-time sensor data updates
def update_sensor_data():
    for i in range(100):
        sensor_data.append(random.randint(0, 100))
        time.sleep(0.1)  # New reading every 100ms

# Start updating data in background
thread = threading.Thread(target=update_sensor_data)
thread.daemon = True
thread.start()

print("Real-time sensor data available at key: sensor_readings")
```

--------------------------------

### Serve ML Model Predictions with znsocket

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Shows how to train a scikit-learn RandomForestClassifier, expose its metadata, predictions, and prediction probabilities via znsocket. It uses DictAdapter and ListAdapter to serve different types of data and a background thread to simulate real-time predictions.

```python
import znsocket
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification

# Train a simple model
X, y = make_classification(n_samples=1000, n_features=20, n_classes=2, random_state=42)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)

# Create model metadata and results storage
model_info = {
    "model_type": "RandomForestClassifier",
    "n_features": 20,
    "n_classes": 2,
    "accuracy": model.score(X, y),
    "feature_importance": model.feature_importances_.tolist()
}

predictions = []
prediction_probabilities = []

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose model information
info_adapter = znsocket.DictAdapter(
    key="model_info",
    socket=client,
    object=model_info
)

# Expose predictions
pred_adapter = znsocket.ListAdapter(
    key="predictions",
    socket=client,
    object=predictions
)

# Expose prediction probabilities
prob_adapter = znsocket.ListAdapter(
    key="prediction_probabilities",
    socket=client,
    object=prediction_probabilities
)

# Simulate real-time predictions
def make_predictions():
    import time
    import random

    while True:
        # Generate random input
        sample = np.random.rand(1, 20)

        # Make prediction
        pred = model.predict(sample)[0]
        prob = model.predict_proba(sample)[0].tolist()

        # Store results
        predictions.append(int(pred))
        prediction_probabilities.append(prob)

        # Update model info
        model_info["total_predictions"] = len(predictions)

        time.sleep(2)  # New prediction every 2 seconds

# Start prediction service
import threading
thread = threading.Thread(target=make_predictions)
thread.daemon = True
thread.start()

print("ML model serving started!")
print("Access model info at: model_info")
print("Access predictions at: predictions")
print("Access probabilities at: prediction_probabilities")
```

--------------------------------

### Python: Complex Nested Adapters

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

This snippet illustrates the creation of complex, nested data structures using znsocket's adapters. It defines a 'user_profiles' dictionary with nested dictionaries and lists, suitable for exposure via a DictAdapter.

```python
import znsocket

# Create complex nested data structure
user_profiles = {
    "user1": {
        "name": "Alice",
        "scores": [85, 92, 78, 95],
        "preferences": {"theme": "dark", "notifications": True}
    },
    "user2": {
        "name": "Bob",
        "scores": [78, 85, 90, 88],
        "preferences": {"theme": "light", "notifications": False}
    }
}

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Expose user profiles through adapter
adapter = znsocket.DictAdapter(
    key="user_profiles",
    socket=client,
    object=user_profiles
)

print("Complex nested user profiles exposed through adapter")
```

--------------------------------

### Python: Cross-Language Copy-on-Write Setup

Source: https://github.com/zincware/znsocket/blob/main/docs/source/copy_on_write.rst

This Python code sets up an original dataset using ZNSocket Lists and Dicts, then demonstrates copy-on-write behavior by modifying a segment. It creates a list of dictionaries, stores them, and then uses Segments to create a modifiable view, ensuring the original data remains untouched.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Create original dataset with Dict objects
lst = znsocket.List(r=client, key="test:data")

data = [
    {"value": [1, 2, 3]},
    {"value": [4, 5, 6]},
    {"value": [7, 8, 9]},
    {"value": [10, 11, 12]},
]

# Use pipeline for efficient batch operations
p = client.pipeline()
msg = []
for idx, value in enumerate(data):
    atoms_dict = znsocket.Dict(r=p, key=f"test:data/{idx}")
    for k, v in value.items():
        atoms_dict[k] = v
    msg.append(atoms_dict)
p.execute()
lst.extend(msg)

# Create copy-on-write view using Segments
segments = znsocket.Segments(r=client, origin=lst, key="test:data/segments")

# Modify a single element (copy-on-write)
value_to_modify = segments[2]
modified_value = value_to_modify.copy("test:data/segments/2")
modified_value["value"] = [100, 200, 300]
segments[2] = modified_value

# Original list remains unchanged: lst[2]["value"] == [7, 8, 9]
# Segments shows modification: segments[2]["value"] == [100, 200, 300]
```

--------------------------------

### Python Example with Large Data Transfer

Source: https://github.com/zincware/znsocket/blob/main/README.md

Demonstrates sending and receiving large data, such as a NumPy array, using ZnSocket's Dict adapter. The code shows how ZnSocket automatically handles chunking and compression for large datasets without requiring explicit user intervention.

```python
import numpy as np
from znsocket import Client, Dict

client = Client.from_url("znsocket://127.0.0.1:5000")

large_data = np.random.rand(1000, 1000)

data_dict = Dict(r=client, key="large_dataset")
data_dict["array"] = large_data

retrieved_data = data_dict["array"]
```

--------------------------------

### Efficiently Handle Large Data with Segments in Python

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

Shows how to use ZNSocket Segments in Python for efficient handling of large datasets. It demonstrates creating a large list, converting it to segments, and accessing data through the segments object, including nesting segments within Dict and List structures.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")

# Create large original list
large_list = znsocket.List(client, "large_data")
large_list.extend(range(1000000))  # 1 million items

# Create segments for efficient copying
segments = znsocket.Segments.from_list(large_list, "data_segments")

# Segments behave like lists but with efficient storage
print(f"Segments length: {len(segments)}")
print(f"First item: {segments[0]}")
print(f"Last item: {segments[-1]}")

# Segments can also be stored in nested structures
data_container = znsocket.Dict(client, "data_container")
data_container["original"] = large_list
data_container["segments"] = segments  # Segments in Dict
data_container["metadata"] = {"size": len(segments), "type": "segments"}

# Access segments through nested structure
stored_segments = data_container["segments"]
print(f"Accessed through dict: {stored_segments[0]}")

# Create a list of different segments
segments_collection = znsocket.List(client, "segments_collection")
segments_collection.append(segments)  # Segments in List

# All data structures (List, Dict, Segments) can be nested within each other
# This provides maximum flexibility for complex data organization
```

--------------------------------

### JavaScript Client for Browser and Node.js

Source: https://context7.com/zincware/znsocket/llms.txt

Provides examples of using the znsocket JavaScript/TypeScript client in both browser and Node.js environments. It covers common operations for Lists and Dicts, including asynchronous methods, real-time refresh callbacks, and async iteration. This enables seamless data sharing across different JavaScript platforms.

```javascript
import { createClient, List, Dict } from 'znsocket';

// Create client and connect
const client = createClient({ url: 'http://127.0.0.1:5000' });
await client.connect();

// Work with Lists
const items = new List({ client, key: 'shopping_cart' });

// All operations are async
await items.push('apple');
await items.push('banana');
console.log(await items.length());     // 2
console.log(await items.get(0));       // "apple"
console.log(await items.slice(0, 2));  // ["apple", "banana"]

// Set items by index
await items.set(0, 'apricot');

// Clear the list
await items.clear();

// Work with Dicts
const user = new Dict({ client, key: 'user:profile' });

await user.set('name', 'Alice');
await user.set('email', 'alice@example.com');

console.log(await user.get('name'));       // "Alice"
console.log(await user.keys());            // ["name", "email"]
console.log(await user.values());          // ["Alice", "alice@example.com"]
console.log(await user.entries());         // [["name", "Alice"], ...]
console.log(await user.length());          // 2

// Bulk update
await user.update({ city: 'Berlin', country: 'Germany' });

// Convert to plain JavaScript object
const userObj = await user.toObject();
console.log(userObj);  // { name: "Alice", email: "...", city: "Berlin", ... }

// Real-time refresh callbacks
items.onRefresh((data) => {
    console.log('List changed:', data);
    // data.start: index where changes started
    // data.indices: specific indices that changed
});

user.onRefresh((data) => {
    console.log('Dict changed:', data);
    // data.keys: array of keys that changed
});

// Clean up listeners
items.offRefresh();
user.offRefresh();

// Async iteration over lists
for await (const item of items) {
    console.log(item);
}

// Disconnect when done
client.disconnect();
```

--------------------------------

### Python: A/B Testing Configurations with ZnSocket Lists

Source: https://github.com/zincware/znsocket/blob/main/docs/source/copy_on_write.rst

Explains how to set up A/B testing and experimentation using ZnSocket Lists. It shows how to define a base configuration and create multiple variants (e.g., variant_a, variant_b) that can fall back to the base configuration.

```python
# Base configuration
base_config = znsocket.List(r=client, key="base_config")

# Create test variants
variant_a = znsocket.List(
    r=client,
    key="variant_a",
    fallback="base_config",
    fallback_policy="frozen"
)

variant_b = znsocket.List(
    r=client,
    key="variant_b",
    fallback="base_config",
    fallback_policy="frozen"
)
```

--------------------------------

### Nested Data Structures: List in Dict (Python)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Demonstrates how to store a znsocket List within a znsocket Dict in Python, allowing for complex, nested data organization and access.

```python
# Python: Store a List inside a Dict
user_data = znsocket.Dict(client, "user_data")
user_scores = znsocket.List(client, "user_scores")
user_scores.extend([85, 92, 78, 95])

user_data["name"] = "John"
user_data["scores"] = user_scores  # List inside Dict

# Access nested data
print(user_data["scores"][0])  # 85
```

--------------------------------

### Nested Data Structures: Dict in List (JavaScript)

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

Shows how to embed a znsocket Dict within a znsocket List in JavaScript, enabling the creation of hierarchical data structures accessible across clients.

```javascript
// JavaScript: Store a Dict inside a List
const users = new List({ client, key: 'users' });
const user1 = new Dict({ client, key: 'user1' });

await user1.set('name', 'Alice');
await user1.set('age', 30);
await users.push(user1);  // Dict inside List

// Access nested data
const firstUser = await users.get(0);
const userName = await firstUser.get('name');  // 'Alice'
```

--------------------------------

### Access Python Data with ZNSocket JavaScript Client

Source: https://github.com/zincware/znsocket/blob/main/docs/source/quickstart.rst

This snippet demonstrates connecting to a ZNSocket server using JavaScript, accessing Python data structures (List and Dict), and performing common operations like retrieving individual items, specific keys, and data slices. It highlights the seamless cross-language data access provided by ZNSocket.

```javascript
import { createClient, List, Dict } from 'znsocket';

const client = createClient({ url: 'http://localhost:5000' });
await client.connect();

// Access Python data from JavaScript
const temperatures = new List({ client, key: 'temperature_readings' });
const settings = new Dict({ client, key: 'app_settings' });

// All operations work seamlessly
console.log(`Latest temp: ${await temperatures.get(-1)}°C`);
console.log(`Database: ${await settings.get('database_url')}`);

// Efficient slicing
const recent = await temperatures.slice(-5);  // Last 5 readings
console.log(`Recent readings: ${recent}`);
```

--------------------------------

### Python Service Consumer for Shared Configuration

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

A Python script that consumes shared configuration from a znsocket server. It reads values from the 'app_config' dictionary and also sets up a listener for configuration changes.

```python
import znsocket

client = znsocket.Client("http://localhost:5000")
config = znsocket.Dict(client, "app_config")

# Read configuration
db_url = config["database_url"]
cache_enabled = config["cache_enabled"]

print(f"Database URL: {db_url}")
print(f"Cache enabled: {cache_enabled}")

# Listen for configuration changes
def on_config_change(data):
    print(f"Configuration changed: {data}")

config.on_refresh(on_config_change)
```

--------------------------------

### ZNSocket Server API

Source: https://github.com/zincware/znsocket/blob/main/docs/source/python_api.rst

Documentation for the `znsocket.Server` class, detailing its methods and inheritance.

```APIDOC
## znsocket.Server

### Description
Provides server-side functionalities for managing ZNSocket connections and services.

### Class
znsocket.Server

### Members
This class has various members detailing its capabilities. Refer to the full Python documentation for specific method signatures and descriptions.

### Inheritance
Inherits from relevant base classes. Refer to the full Python documentation for details.
```

--------------------------------

### Run Python Tests with Docker and Uvicorn

Source: https://github.com/zincware/znsocket/blob/main/docs/source/develop.rst

This snippet details how to run Python tests for ZNSocket. It requires a Redis instance running on port 6379 and uses `uv` with `pytest` to execute the tests.

```bash
docker run --rm -p 6379:6379 redis
uv run pytest tests
```

--------------------------------

### Python Client for Real-time Chat

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

A Python client for a real-time chat application. It connects to a znsocket server, manages a shared list of messages, listens for new messages, and allows users to send messages.

```python
import znsocket
import threading

# Connect to server
client = znsocket.Client("http://localhost:5000")

# Create shared message list
messages = znsocket.List(client, "chat_messages")

# Listen for new messages
def on_message_added(data):
    if 'indices' in data:
        for idx in data['indices']:
            print(f"New message: {messages[idx]}")

messages.on_refresh(on_message_added)

# Send messages
while True:
    message = input("Enter message: ")
    messages.append(f"User: {message}")
```

--------------------------------

### Python Dict and List with Fallback Keys

Source: https://context7.com/zincware/znsocket/llms.txt

Explains the use of fallback keys in Python's Dict and List classes for layered data access, useful for defaults or versioned configurations. It demonstrates how to create configurations with fallbacks and how 'frozen' and 'copy' policies affect data retrieval and storage. This enables flexible configuration management.

```python
from znsocket import Client, List, Dict

client = Client.from_url("znsocket://127.0.0.1:5000")

# Create default configuration
defaults = Dict(r=client, key="config:defaults")
defaults["theme"] = "light"
defaults["language"] = "en"
defaults["notifications"] = True

# Create user config with fallback to defaults
user_config = Dict(
    r=client,
    key="config:user:123",
    fallback="config:defaults",
    fallback_policy="frozen"  # Read from fallback, don't copy
)

# User config is empty, reads from fallback
print(user_config["theme"])        # "light" (from defaults)
print(user_config["language"])     # "en" (from defaults)

# User overrides are stored in user config
user_config["theme"] = "dark"
print(user_config["theme"])        # "dark" (user override)
print(user_config["language"])     # "en" (still from defaults)

# With "copy" policy, fallback data is copied on first access
user_config_copy = Dict(
    r=client,
    key="config:user:456",
    fallback="config:defaults",
    fallback_policy="copy"  # Copy fallback data to this key
)

# Same pattern works for Lists
default_items = List(r=client, key="items:defaults")
default_items.extend(["item1", "item2", "item3"])

user_items = List(
    r=client,
    key="items:user:123",
    fallback="items:defaults",
    fallback_policy="frozen"
)
```

--------------------------------

### ZNSocket Client API

Source: https://github.com/zincware/znsocket/blob/main/docs/source/python_api.rst

Documentation for the `znsocket.Client` class, detailing its methods and inheritance.

```APIDOC
## znsocket.Client

### Description
Provides client-side functionalities for interacting with ZNSocket services.

### Class
znsocket.Client

### Members
This class has various members detailing its capabilities. Refer to the full Python documentation for specific method signatures and descriptions.

### Inheritance
Inherits from relevant base classes. Refer to the full Python documentation for details.
```

--------------------------------

### Python Task Producer for Distributed Queue

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

A Python script that acts as a task producer for a distributed task queue. It connects to a znsocket server and appends tasks (as JSON strings) to a shared list named 'task_queue'.

```python
import znsocket
import json

client = znsocket.Client("http://localhost:5000")
task_queue = znsocket.List(client, "task_queue")

# Add tasks to queue
tasks = [
    {"id": 1, "type": "process_data", "data": "sample1"},
    {"id": 2, "type": "process_data", "data": "sample2"},
    {"id": 3, "type": "send_email", "to": "user@example.com"},
]

for task in tasks:
    task_queue.append(json.dumps(task))

print(f"Added {len(tasks)} tasks to queue")
```

--------------------------------

### Python: Data Preprocessing Pipelines with ZnSocket Segments

Source: https://github.com/zincware/znsocket/blob/main/docs/source/copy_on_write.rst

Demonstrates building data preprocessing pipelines using ZnSocket. It shows how to create different variants of raw data (e.g., normalized, filtered) using Segments and apply transformations selectively to these variants.

```python
# Raw dataset
raw_data = znsocket.List(r=client, key="raw_data")

# Create preprocessing variants
normalized_data = znsocket.Segments(r=client, origin=raw_data, key="normalized")
filtered_data = znsocket.Segments(r=client, origin=raw_data, key="filtered")

# Apply transformations only where needed
for i, sample in enumerate(raw_data):
    if sample["quality_score"] < threshold:
        cleaned_sample = preprocess(sample)
        filtered_data[i] = cleaned_sample
```

--------------------------------

### Programmatic Server Creation (Python)

Source: https://context7.com/zincware/znsocket/llms.txt

Creates and runs a ZnSocket server programmatically in Python with custom settings for port, buffer size, storage backend, and logging.

```python
from znsocket import Server

# Programmatic server creation with custom settings
server = Server(
    port=5000,
    max_http_buffer_size=10 * 1024 * 1024,  # 10MB buffer
    storage="memory",  # or "redis://localhost:6379"
    logger=True
)

# Run the server (blocking)
server.run()
```

--------------------------------

### Python Task Consumer for Distributed Queue

Source: https://github.com/zincware/znsocket/blob/main/docs/source/examples.rst

A Python script that acts as a task consumer for a distributed task queue. It connects to a znsocket server, retrieves tasks from the 'task_queue' list, processes them, and simulates work using time.sleep.

```python
import znsocket
import json
import time

client = znsocket.Client("http://localhost:5000")
task_queue = znsocket.List(client, "task_queue")

def process_task(task_data):
    task = json.loads(task_data)
    print(f"Processing task {task['id']}: {task['type']}")
    time.sleep(1)  # Simulate work
    print(f"Task {task['id']} completed")

# Process tasks
while True:
    if len(task_queue) > 0:
        task = task_queue.pop(0)  # Get first task
        process_task(task)
    else:
        time.sleep(0.1)  # Wait for new tasks
```

--------------------------------

### ZNSocket ListAdapter API

Source: https://github.com/zincware/znsocket/blob/main/docs/source/python_api.rst

Documentation for the `znsocket.ListAdapter` class, detailing its methods and inheritance.

```APIDOC
## znsocket.ListAdapter

### Description
An adapter for list-like objects within the ZNSocket framework.

### Class
znsocket.ListAdapter

### Members
This class has various members detailing its capabilities. Refer to the full Python documentation for specific method signatures and descriptions.

### Inheritance
Inherits from relevant base classes. Refer to the full Python documentation for details.
```

--------------------------------

### ZNSocket List API

Source: https://github.com/zincware/znsocket/blob/main/docs/source/python_api.rst

Documentation for the `znsocket.List` class, detailing its methods and inheritance.

```APIDOC
## znsocket.List

### Description
Represents a list data structure with ZNSocket-specific functionalities.

### Class
znsocket.List

### Members
This class has various members detailing its capabilities. Refer to the full Python documentation for specific method signatures and descriptions.

### Inheritance
Inherits from relevant base classes. Refer to the full Python documentation for details.
```
Tokens:
Raw