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")

Public Downloads (no access token)

Consumers who only need to download a published file don't need user credentials. Create the client without an access token and download by the file's public shortcut:

# No access token required for public files
client = StorageClient(base_url="https://h2-storage.kamimind.ai")

# Resolve the public file's details (presigned URL, name, size, ...)
info = client.get_public_download_info("abc12345")
print(info.file_name, info.file_size)

# Download to a file
path = client.download_public_file("abc12345", "downloads/file.pdf")

# Or download as bytes
content = client.download_public_bytes("abc12345")

AsyncStorageClient exposes the same get_public_download_info, download_public_file, and download_public_bytes methods.

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}")

Debug Logging

The SDK logs through the standard logging module under the kamimind_storage logger and configures nothing itself, so it is silent unless you opt in:

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("kamimind_storage").setLevel(logging.DEBUG)

Failed requests are logged at WARNING without any opt-in. Turning on DEBUG adds the values that vary from one file to the next, which is what you need when an upload fails for one file while its siblings succeed:

DEBUG  upload initiate: filename='projects/14/slides/slide_009.png'
       sanitized='projects_14_slides_slide_009.png' size=48 content_type='image/png'
       multipart=False is_public=False metadata={'slide': 9, 'title': 'Tổng quan – Q3'}
DEBUG  presigned PUT: bytes=48 content_type='image/png' url=https://...slide_009.png?<redacted>
WARN   request failed: PUT https://...slide_009.png?<redacted> status=500
       upstream_ids={'cf-ray': '8f2a...-SIN', 'x-amz-request-id': '1740...'}
       body=<Error><Code>InternalError</Code>...

Two things are worth knowing:

• Presigned URLs are redacted. Only the scheme, host, and object path are logged; the X-Amz-Signature and X-Amz-Credential query parameters are stripped, so a log file never carries upload credentials.
• upstream_ids is what support needs. cf-ray and x-amz-request-id identify the request inside Cloudflare or S3. Quote them when reporting a server-side InternalError — they are not recoverable from the exception.

Because metadata is logged verbatim at DEBUG, keep that level off in environments where metadata carries personal data.

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