Kamimind Storage SDK for Python

Official Python SDK for Kamimind Storage Service.

Installation

Install from PyPI:

# Using uv (recommended)
uv add kamimind-storage

# Using pip
pip install kamimind-storage

Quick Start

from kamimind_storage import StorageClient

# Initialize client
client = StorageClient(
    base_url="https://api.kamimind.ai",
    access_token="your-jwt-token",
)

# Upload a file
session = client.upload_file("path/to/file.pdf", is_public=True)
print(f"Uploaded: {session.session_id}")

# Get download URL
download = client.get_download_url(session.session_id)
print(f"Download URL: {download.download_url}")

# Make file public
public_info = client.make_public(session.session_id)
print(f"Public URL: {public_info.public_url}")

Async Support

from kamimind_storage import AsyncStorageClient

async def main():
    async with AsyncStorageClient(
        base_url="https://api.kamimind.ai",
        access_token="your-jwt-token",
    ) as client:
        session = await client.upload_file("path/to/file.pdf")
        print(f"Uploaded: {session.session_id}")

Features

File Upload

# Upload from file path
session = client.upload_file("path/to/file.pdf")

# Upload from bytes
session = client.upload_bytes(b"Hello, World!", filename="hello.txt")

# Upload with progress callback
def on_progress(progress):
    print(f"Uploaded: {progress.percentage:.1f}%")

session = client.upload_file("large_file.zip", on_progress=on_progress)

# Upload with metadata
session = client.upload_file(
    "document.pdf",
    metadata={"category": "reports", "year": 2024},
    is_public=True,
)

File Download

# Get download URL
download_info = client.get_download_url(session_id)
print(download_info.download_url)

# Download to file
path = client.download_file(session_id, "downloads/file.pdf")

# Download as bytes
content = client.download_bytes(session_id)

Public Sharing

# Make file public
public_info = client.make_public(session_id)
print(f"Public URL: {public_info.public_url}")

# Toggle public access
client.toggle_public_access(session_id, is_public=False)

# Get public URL by shortcut
public_url = client.get_public_url("abc12345")

Session Management

# Get upload status
status = client.get_upload_status(session_id)
print(f"Status: {status.status}")

# Abort upload
client.abort_upload(session_id)

Configuration

client = StorageClient(
    base_url="https://api.kamimind.ai",
    access_token="your-jwt-token",
    timeout=60.0,              # Request timeout in seconds
    max_retries=3,             # Max retry attempts
    multipart_threshold=100 * 1024 * 1024,  # 100MB threshold for multipart
    part_size=5 * 1024 * 1024,  # 5MB part size
)

Error Handling

from kamimind_storage.exceptions import (
    StorageError,
    AuthenticationError,
    NotFoundError,
    UploadError,
    QuotaExceededError,
)

try:
    session = client.upload_file("file.pdf")
except AuthenticationError:
    print("Invalid or expired token")
except QuotaExceededError:
    print("Storage quota exceeded")
except UploadError as e:
    print(f"Upload failed: {e.message}")
except StorageError as e:
    print(f"Storage error: {e.message}")

Development

# Install uv if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create virtual environment and install dependencies
uv sync

# Run tests
uv run pytest

# Run with coverage
uv run pytest --cov=kamimind_storage

# Lint
uv run ruff check .

# Format
uv run ruff format .

# Type check
uv run mypy src

License

MIT License