Arkiv SDK

Arkiv is a permissioned storage system for decentralized apps, supporting flexible entities with binary data, attributes, and metadata.

The Arkiv SDK is the official Python library for interacting with Arkiv networks. It offers a type-safe, developer-friendly API for managing entities, querying data, subscribing to events, and offchain verification—ideal for both rapid prototyping and production use.

Architecture

Principles:

• The SDK is based on a modern and stable client library.
• The SDK should feel like "Library + Entities"

As underlying library we use Web3.py (no good alternatives for Python).

Arkiv Client

The Arkiv SDK should feel like "web3.py + entities", maintaining the familiar developer experience that Python web3 developers expect.

A client.arkiv.* approach is in line with web3.py's module pattern. It clearly communicates that arkiv is a module extension just like eth, net, etc.

Hello World

Synchronous API

Here's a "Hello World!" example showing how to use the Python Arkiv SDK:

fromarkivimportArkiv# Create Arkiv client with default settings:# - starting and connecting to a containerized Arkiv node# - creating a funded default accountclient=Arkiv()
print(f"Client: {client}, connected: {client.is_connected()}")
print(f"Account: {client.eth.default_account}")
print(f"Balance: {client.from_wei(client.eth.get_balance(client.eth.default_account), 'ether')} ETH")
# Create entity with data and attributesentity_key, receipt=client.arkiv.create_entity(
payload=b"Hello World!",
content_type="text/plain",
attributes= {"type": "greeting", "version": 1},
expires_in=client.arkiv.to_seconds(days=1)
)
# Get individual entity and print its detailsentity=client.arkiv.get_entity(entity_key)
print(f"Creation TX: {receipt.tx_hash}")
print(f"Entity: {entity}")

Asynchronous API

For async/await support, use AsyncArkiv:

importasynciofromarkivimportAsyncArkivasyncdefmain():
# Create async client with default settingsasyncwithAsyncArkiv() asclient:
# Create entity with data and attributesentity_key, tx_hash=awaitclient.arkiv.create_entity(
payload=b"Hello Async World!",
content_type="text/plain",
attributes= {"type": "greeting", "version": 1},
expires_in=client.arkiv.to_seconds(days=1)
)
# Get entity and check existenceentity=awaitclient.arkiv.get_entity(entity_key)
exists=awaitclient.arkiv.entity_exists(entity_key)
asyncio.run(main())

Web3 Standard Support

fromweb3importHTTPProviderprovider=HTTPProvider('https://mendoza.hoodi.arkiv.network/rpc')
# Arkiv 'is a' Web3 clientclient=Arkiv(provider)
balance=client.eth.get_balance(client.eth.default_account)
tx=client.eth.get_transaction(tx_hash)

Entity Operations

Beyond creating and reading entities, Arkiv supports updating, extending, transferring ownership, and deleting entities.

Update Entity

Modify an entity's payload, attributes, or expiration:

# Update entity with new payload and attributesentity_key, receipt=client.arkiv.update_entity(
entity_key,
payload=b"Updated content",
attributes={"type": "greeting", "version": 2},
expires_in=client.arkiv.to_seconds(days=7)
)

Extend Entity Lifetime

Extend an entity's expiration without modifying its content:

# Extend entity lifetime by 30 daysentity_key, receipt=client.arkiv.extend_entity(
entity_key,
extend_by=client.arkiv.to_seconds(days=30)
)

Change Entity Owner

Transfer ownership of an entity to another address:

# Transfer entity to a new ownernew_owner="0x1234567890abcdef1234567890abcdef12345678"entity_key, receipt=client.arkiv.change_owner(entity_key, new_owner)

Delete Entity

Permanently remove an entity (only the owner can delete):

# Delete entityreceipt=client.arkiv.delete_entity(entity_key)

Advanced Features

Query Builder

The query builder provides a clean, chainable API for querying entities. It wraps the lower-level query_entities method with a SQL-like interface.

Basic Usage

fromarkivimportArkiv, StrAttrfromarkiv.typesimportKEY, ATTRIBUTESclient=Arkiv()
# Define typed attributesentity_type=StrAttr("type")
status=StrAttr("status")
# Simple query - select all fieldsresults=client.arkiv.select().where(entity_type=="user").fetch()
forentityinresults:
print(f"Entity: {entity.key}")
# Select specific fieldsresults=client.arkiv.select(KEY, ATTRIBUTES).where(status=="active").fetch()
# Count matching entitiescount=client.arkiv.select().where(entity_type=="user").count()
print(f"Found {count} users")

Expressions with IntAttr/StrAttr

The expression builder generates SQL-like query strings under the hood, providing a type-safe Python API for constructing filter conditions. The .where() method accepts either an Expr object from the expression builder or a raw SQL-like query string (see Query Language below).

For dynamic query building with runtime type checking, use the expression builder:

fromarkivimportArkiv, IntAttr, StrAttr, IntSort, DESCclient=Arkiv()
# Define typed attributesage=IntAttr("age")
status=StrAttr("status")
role=StrAttr("role")
# Build expressions with operatorsresults=client.arkiv.select() \
.where((age>=18) & (status=="active")) \
.order_by(IntSort("age", DESC)) \
.fetch()
# Complex expressions with OR and ANDresults=client.arkiv.select() \
.where((role=="admin") | (role=="moderator") & (status=="active")) \
.fetch()
# NOT operatorresults=client.arkiv.select() \
.where((age>=18) &~(status=="banned")) \
.fetch()
# Type checking catches errors earlyage=="18"# TypeError: IntAttr 'age' requires int, got strstatus==1# TypeError: StrAttr 'status' requires str, got int

Expression Operators:

• & - AND
• | - OR
• ~ - NOT

Note: Always use parentheses around comparisons when combining with &, |, or ~ due to Python operator precedence.

Sorting with IntSort/StrSort

Use type-specific sort classes for ORDER BY clauses:

fromarkivimportArkiv, IntSort, StrSort, StrAttr, DESCclient=Arkiv()
# Define typed attributesentity_type=StrAttr("type")
status=StrAttr("status")
# Define sorting_criteriastatus_asc=StrSort("status")
age_desc=IntSort("age", DESC)
# Sort by age descendingresults=client.arkiv.select() \
.where(entity_type=="user") \
.order_by(age_desc) \
.fetch()
# Multi-field sorting: status ascending, then age descendingresults=client.arkiv.select() \
.where(status=="active") \
.order_by(status_asc, age_desc) \
.fetch()

Limiting Results

Use .limit() to restrict the total number of results and .max_page_size() to control pagination:

fromarkivimportArkiv, IntSort, StrAttr, DESCclient=Arkiv()
# Define typed attributeentity_type=StrAttr("type")
# Get first 10 matching entitiesresults=client.arkiv.select() \
.where(entity_type=="user") \
.limit(10) \
.fetch()
# Top 5 users by ageresults=client.arkiv.select() \
.where(entity_type=="user") \
.order_by(IntSort("age", DESC)) \
.limit(5) \
.fetch()
# Control page size for large entities (smaller pages = less memory per request)results=client.arkiv.select() \
.where(entity_type=="document") \
.max_page_size(10) \
.fetch()
# Combine limit and page sizeresults=client.arkiv.select() \
.where(entity_type=="user") \
.limit(100) \
.max_page_size(25) \
.fetch()

Batch Operations

Batch operations allow you to group multiple entity operations (create, update, extend, delete, change_owner) into a single atomic transaction. This is more efficient and ensures all operations either succeed or fail together.

Basic Usage

fromarkivimportArkivclient=Arkiv()
# Using context manager (recommended)withclient.arkiv.batch() asbatch:
batch.create_entity(payload=b"item 1", expires_in=3600)
batch.create_entity(payload=b"item 2", expires_in=3600)
batch.create_entity(payload=b"item 3", expires_in=3600)
# Batch is automatically executed on exitprint(f"Created {len(batch.receipt.creates)} entities")
# Access created entity keysforcreate_eventinbatch.receipt.creates:
print(f"Created: {create_event.key}")

Loop-Based Creation

Batch operations work naturally with loops:

items= [
{"name": "alice", "role": "admin"},
{"name": "bob", "role": "user"},
{"name": "charlie", "role": "user"},
]
withclient.arkiv.batch() asbatch:
foriteminitems:
batch.create_entity(
payload=item["name"].encode(),
attributes={"role": item["role"]},
expires_in=3600,
)
print(f"Created {len(batch.receipt.creates)} users")

Mixed Operations

A single batch can contain different operation types:

withclient.arkiv.batch() asbatch:
# Create new entitiesbatch.create_entity(payload=b"new item", expires_in=3600)
# Update existing entitiesbatch.update_entity(existing_key, payload=b"updated", expires_in=3600)
# Extend entity lifetimebatch.extend_entity(another_key, extend_by=7200)
# Change ownershipbatch.change_owner(some_key, new_owner_address)
# Delete entitiesbatch.delete_entity(old_key)
# Check resultsprint(f"Creates: {len(batch.receipt.creates)}")
print(f"Updates: {len(batch.receipt.updates)}")
print(f"Extensions: {len(batch.receipt.extensions)}")
print(f"Deletes: {len(batch.receipt.deletes)}")

Manual Execution

For more control, you can execute batches manually:

batch=client.arkiv.batch()
batch.create_entity(payload=b"data", expires_in=3600)
batch.create_entity(payload=b"more data", expires_in=3600)
# Execute explicitlyreceipt=batch.execute()
print(f"Transaction: {receipt.tx_hash}")

Async Support

Batch operations work with AsyncArkiv:

asyncwithAsyncArkiv() asclient:
asyncwithclient.arkiv.batch() asbatch:
batch.create_entity(payload=b"async item 1", expires_in=3600)
batch.create_entity(payload=b"async item 2", expires_in=3600)
print(f"Created {len(batch.receipt.creates)} entities")

Error Handling

• If an exception occurs inside the context manager, the batch is not executed
• Empty batches are silently skipped (no-op)
• All operations in a batch are atomic: if any operation fails, the entire batch is rolled back

try:
withclient.arkiv.batch() asbatch:
batch.create_entity(payload=b"item 1", expires_in=3600)
raiseValueError("Something went wrong")
batch.create_entity(payload=b"item 2", expires_in=3600)
exceptValueError:
pass# Batch was not executed - no entities createdassertbatch.receiptisNone

Provider Builder

The ProviderBuilder provides a fluent API for creating providers to connect to various Arkiv networks:

fromarkivimportArkivfromarkiv.accountimportNamedAccountfromarkiv.providerimportProviderBuilder# Create account from wallet jsonwithopen('wallet_bob.json', 'r') asf:
wallet=f.read()
bob=NamedAccount.from_wallet('Bob', wallet, 's3cret')
# Initialize Arkiv client connected to Kaolin (Arkiv testnet)provider=ProviderBuilder().kaolin().build()
client=Arkiv(provider, account=bob)
# Additional builder examplesprovider_custom=ProviderBuilder().custom("https://mendoza.hoodi.arkiv.network/rpc").build()
provider_container=ProviderBuilder().node().build()
provider_kaolin_ws=ProviderBuilder().kaolin().ws().build()

Query Language

Arkiv uses a SQL-like query language to filter and retrieve entities based on their attributes. The query language supports standard comparison operators, logical operators, and parentheses for complex conditions.

Supported Operators

Comparison Operators:

• = - Equal to
• != - Not equal to
• > - Greater than
• >= - Greater than or equal to
• < - Less than
• <= - Less than or equal to

Logical Operators:

• AND - Logical AND
• OR - Logical OR
• NOT - Logical NOT (can also use !=)

Parentheses can be used to group conditions and control evaluation order.

Query Examples

fromarkivimportArkivclient=Arkiv()
# Simple equalityquery='type = "user"'entities=list(client.arkiv.query_entities(query))
# Note that inn the examples below the call to query_entities is omitted# Multiple conditions with ANDquery='type = "user" AND status = "active"'# OR conditions with parenthesesquery='type = "user" AND (status = "active" OR status = "pending")'# Comparison operatorsquery='type = "user" AND age >= 18 AND age < 65'# NOT conditionsquery='type = "user" AND status != "deleted"'# Alternative NOT syntaxquery='type = "user" AND NOT (status = "deleted")'# Complex nested conditionsquery='(type = "user" OR type = "admin") AND (age >= 18 AND age <= 65)'# Multiple NOT conditionsquery='type = "user" AND status != "deleted" AND status != "banned"'# Pattern matching with GLOB (using * as wildcard)query='name GLOB "John*"'# Names starting with "John"# Pattern matching with suffixquery='email GLOB "*@example.com"'# Emails ending with @example.com

Note: String values in queries must be enclosed in double quotes ("). Numeric values do not require quotes. The GLOB operator supports pattern matching using * as a wildcard character. Note that the GLOB operator might be replace by a SQL standard LIKE operator in the future.

Watch Entity Events

Arkiv provides near real-time event monitoring for entity lifecycle changes. You can watch for entity creation, updates, extensions, deletions, and ownership changes using callback-based event filters.

Available Event Types

• watch_entity_created - Monitor when new entities are created
• watch_entity_updated - Monitor when entities are updated
• watch_entity_extended - Monitor when entity lifetimes are extended
• watch_entity_deleted - Monitor when entities are deleted
• watch_owner_changed - Monitor when entity ownership changes

Basic Usage

fromarkivimportArkivclient=Arkiv()
# Define callback function to handle eventsdefon_entity_created(event, tx_hash):
print(f"New entity created: {event.key}")
print(f"Owner: {event.owner}")
print(f"Transaction: {tx_hash}")
# Start watching for entity creation eventsevent_filter=client.arkiv.watch_entity_created(on_entity_created)
# Create an entity - callback will be triggeredentity_key, _=client.arkiv.create_entity(
payload=b"Hello World",
attributes={"type": "greeting"}
)
# Stop watching when doneevent_filter.stop()
event_filter.uninstall()

Watching Multiple Event Types

created_events= []
updated_events= []
deleted_events= []
defon_created(event, tx_hash):
created_events.append((event, tx_hash))
defon_updated(event, tx_hash):
updated_events.append((event, tx_hash))
defon_deleted(event, tx_hash):
deleted_events.append((event, tx_hash))
# Watch multiple event types simultaneouslyfilter_created=client.arkiv.watch_entity_created(on_created)
filter_updated=client.arkiv.watch_entity_updated(on_updated)
filter_deleted=client.arkiv.watch_entity_deleted(on_deleted)
# Perform operations...# Events are captured in real-time# Cleanup all filtersfilter_created.uninstall()
filter_updated.uninstall()
filter_deleted.uninstall()

Historical Events

By default, watchers only capture new events from the current block forward. You can also watch from a specific historical block:

# Watch from a specific block numberevent_filter=client.arkiv.watch_entity_created(
on_entity_created,
from_block=1000
)
# Watch from the beginning of the chainevent_filter=client.arkiv.watch_entity_created(
on_entity_created,
from_block=0
)

Automatic Cleanup

When using Arkiv as a context manager, all event filters are automatically cleaned up on exit:

withArkiv() asclient:
# Create event filtersfilter1=client.arkiv.watch_entity_created(callback1)
filter2=client.arkiv.watch_entity_updated(callback2)
# Perform operations...# Filters are automatically stopped and uninstalled when exiting context

You can also manually clean up all active filters:

client.arkiv.cleanup_filters()

Note: Event watching requires polling the node for new events. The SDK handles this automatically in the background.

Development Guide

Setup

Requirements

• Python: Version 3.10 or higher
• Install:
  • pip install --pre arkiv-sdk
  • pip install testcontainers websockets
• RPC: https://mendoza.hoodi.arkiv.network/rpc

Branches, Versions, Changes

Branches

The current stable branch on Git is main. Currently main hosts the initial SDK implementation.

The branch v1-dev hosts the future V1.0 SDK release.

Versions

For version management the uv package and project manger is used. Use the command below to display the current version

uv version

SDK versions are tracked in the following files:

• pyproject.toml
• uv.lock

Testing

Pytest is used for unit and integration testing.

uv run pytest # Run all tests
uv run pytest -k test_create_entity_simple --log-cli-level=info # Specific tests via keyword, print at info log level

If an .env file is present the unit tests are run against the specifice RPC coordinates and test accounts. An example wallet file is provided in .env.testing Make sure that the specified test accounts are properly funded before running the tests.

Otherwise, the tests are run against a testcontainer containing an Arkiv RPC Node. Test accounts are created on the fly and using the CLI inside the local RPC Nonde.

Account wallets for such tests can be created via the command shown below. The provided example creates the wallet file wallet_alice.json using the password provided during the execution of the command.

uv run python -m arkiv.account alice

Code Quality

This project uses comprehensive unit testing, linting and type checking to maintain high code quality:

Quick Commands

Before any commit run quality checks:

./scripts/check-all.sh

Tools Used

• MyPy: Static type checker with strict configuration
• Ruff: Fast linter and formatter (replaces black, isort, flake8, etc.)
• Pre-commit: Automated quality checks on git commits

Individual commands

uv run ruff check . --fix # Lint and auto-fix
uv run ruff format .# Format code
uv run mypy src/ tests/ # Type check
uv run pytest tests/ -v # Run tests
uv run pytest --cov=src # Run code coverage
uv run pre-commit run --all-files # Manual pre commit checks

Pre-commit Hooks

Pre-commit hooks run automatically on git commit and will:

• Fix linting issues with ruff
• Format code consistently
• Run type checking with mypy
• Check file formatting (trailing whitespace, etc.)

MyPy Settings

• strict = true - Enable all strict checks
• no_implicit_reexport = true - Require explicit re-exports
• warn_return_any = true - Warn about returning Any values
• Missing imports are ignored for third-party libraries without type stubs

Ruff Configuration

• Use 88 character line length (Black-compatible)
• Target Python 3.10+ features
• Enable comprehensive rule sets (pycodestyle, pyflakes, isort, etc.)
• Auto-fix issues where possible
• Format with double quotes and trailing commas

Alias

functiongl { git log --format="%C(green)%ad%C(reset) %C(yellow)%h%C(reset)%C(auto)%d%C(reset) %s" --date=format:"%Y-%m-%d_%H:%M:%S" -n ${1:-10}; }
alias gs='git status'