IIPOS Python Client SDK

Thread-safe Python client for IIPOS distributed blob storage. Provides high-performance access to IIPOS clusters with automatic connection pooling, error handling, and retry logic.

Features

• High Performance: Optimized gRPC client with connection pooling
• Thread-Safe: Safe for concurrent access from multiple threads
• Error Handling: Automatic retries and comprehensive error reporting
• Streaming Support: Efficient large object streaming
• TLS Support: Encrypted communication with IIPOS servers
• Batch Operations: Efficient batch put/get/delete operations
• Metadata Support: Key metadata handling

Installation

From PyPI

pip install iipos-client

From Source

git clone https://github.com/sid19991/iipos.git
cd iipos/clients/python
pip install -e .

Development Installation

pip install -e ".[dev]"

Quick Start

from iipos_client import IIPOSClient

# Connect to IIPOS cluster
client = IIPOSClient("localhost:50051", api_key="my-api-key")

# Put an object
client.put("my-key", b"Hello, World!")

# Get an object
data = client.get("my-key")
print(data)  # Output: b'Hello, World!'

# Delete an object
client.delete("my-key")

# Close connection
client.close()

Usage Examples

Basic Operations

from iipos_client import IIPOSClient

client = IIPOSClient("iipos-node-1:50051", api_key="my-api-key")

# PUT - store data
client.put("users:123", b"user_data_here")

# GET - retrieve data
data = client.get("users:123")

# DELETE - remove data
client.delete("users:123")

# EXISTS - check if key exists
exists = client.exists("users:123")

client.close()

With TLS

client = IIPOSClient(
    "iipos-cluster.example.com:50051",
    api_key="secure-api-key",
    use_tls=True,
    cert_path="/path/to/ca-cert.pem",
    tls_server_name="iipos-cluster.example.com"
)

Large Objects

# Streaming put for large files
with open("large-file.bin", "rb") as f:
    client.stream_put("large-key", f)

# Streaming get for large objects
with open("downloaded.bin", "wb") as f:
    client.stream_get("large-key", f)

Batch Operations

# Batch PUT
items = [
    ("key1", b"data1", {"tier": "hot"}),
    ("key2", b"data2", {"tier": "warm"}),
    ("key3", b"data3", {"tier": "cold"}),
]
results = client.batch_put(items)

# Batch GET
keys = ["key1", "key2", "key3"]
results = client.batch_get(keys)

# Batch DELETE
client.batch_delete(keys)

Cluster Client

For automatic failover and load balancing across cluster nodes:

from iipos_client import IIPOSClusterClient

nodes = [
    "iipos-node-1:50051",
    "iipos-node-2:50051",
    "iipos-node-3:50051",
]

client = IIPOSClusterClient(nodes, api_key=["my-api-key-node1","my-api-key-node2","my-api-key-node3"])

# Automatically handles failover and balancing
client.put("resilient-key", b"data")
data = client.get("resilient-key")

client.close()

With Context Manager

from iipos_client import IIPOSClient

# Automatically closes connection
with IIPOSClient("localhost:50051", api_key="my-api-key") as client:
    client.put("key1", b"value1")
    value = client.get("key1")
    print(value)

Error Handling

from iipos_client import IIPOSClient, IIPOSClientError, IIPOSNotFoundError

client = IIPOSClient("localhost:50051", api_key="my-api-key")

try:
    data = client.get("non-existent-key")
except IIPOSNotFoundError:
    print("Key not found")
except IIPOSClientError as e:
    print(f"IIPOS error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
finally:
    client.close()

Configuration

Connection Parameters

client = IIPOSClient(
    address="localhost:50051",           # Server address
    api_key="my-api-key",                # Authentication key
    use_tls=False,                       # Enable TLS encryption
    cert_path=None,                      # Path to CA certificate
    timeout=30,                          # RPC timeout (seconds)
    tls_server_name=None,                # Override TLS server name
    max_retries=3,                       # Max retry attempts
    backoff_factor=1.0,                  # Exponential backoff multiplier
    connection_pool_size=10,             # Max connections
)

API Reference

IIPOSClient

Methods

• put(key: str, data: bytes, metadata: dict = None, ttl_ms: int = 0) -> None

  • Store data in IIPOS
  • key: Unique identifier
  • data: Binary data to store
  • metadata: Optional metadata dict
  • ttl_ms: Time-to-live in milliseconds (0 = no expiration)

• get(key: str) -> bytes

  • Retrieve data from IIPOS
  • key: Key to retrieve
  • Returns: Binary data
  • Raises: IIPOSNotFoundError if key doesn't exist

• delete(key: str) -> None

  • Delete data from IIPOS
  • key: Key to delete

• exists(key: str) -> bool

  • Check if key exists
  • Returns: True if key exists, False otherwise

• stream_put(key: str, file_obj, metadata: dict = None) -> None

  • Stream large file to IIPOS
  • file_obj: File object opened in binary mode
  • metadata: Optional metadata

• stream_get(key: str, file_obj) -> None

  • Stream large object from IIPOS to file
  • file_obj: File object opened in binary mode for writing

• batch_put(items: List[Tuple]) -> List[PutResult]

  • Batch store multiple items
  • items: List of (key, data, metadata) tuples
  • Returns: List of results with status

• batch_get(keys: List[str]) -> List[GetResult]

  • Batch retrieve multiple items
  • keys: List of keys to retrieve
  • Returns: List of results with data

• batch_delete(keys: List[str]) -> List[DeleteResult]

  • Batch delete multiple items
  • keys: List of keys to delete
  • Returns: List of results with status

• close() -> None

  • Close connection to server

Exception Hierarchy

IIPOSError (base)
├── IIPOSClientError
├── IIPOSServerError
├── IIPOSNotFoundError
├── IIPOSAlreadyExistsError
├── IIPOSInvalidArgumentError
└── IIPOSTimeoutError

Performance Tips

1. Use Batch Operations: Batch operations are more efficient than individual requests
2. Streaming: Use stream_put/stream_get for large objects (> 1MB)
3. Connection Pooling: The client automatically pools connections
4. Error Handling: Implement exponential backoff for retries
5. Cluster Client: Use IIPOSClusterClient for fault tolerance

Testing

# Run tests
pytest

# Run with coverage
pytest --cov=iipos_client

# Run specific test
pytest tests/test_iipos_client.py::test_put_get

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit changes (git commit -m 'Add amazing feature')
4. Push to branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Requirements

• Python 3.8+
• grpcio >= 1.50.0
• protobuf >= 3.20.0

License

This project is licensed under the MIT License - see the LICENSE file for details.

Support

• Documentation: IIPOS Documentation
• Issues: GitHub Issues
• Discussions: GitHub Discussions

Changelog

1.0.0 (2026-06-07)

• Initial release
• Core put/get/delete operations
• Batch operations support
• Streaming support for large objects
• TLS support
• Cluster client with failover
• Comprehensive error handling
• Connection pooling
• Retry logic with exponential backoff