IQLabs SDK (Python)

Draft: This document is in progress and will be refined.

Core Concepts

These are the key concepts to know before using the IQLabs SDK.

Data Storage (Code In)

This is how you store any data (files, text, JSON) on-chain.

How is it stored?

Depending on data size, the SDK picks the optimal method:

Small data (< 700 bytes): store immediately, fastest
Medium data (< 8.5 KB): split into multiple transactions
Large data (>= 8.5 KB): upload in parallel for speed

Key related functions

code_in(): upload data and get a transaction ID
read_code_in(): read data back from a transaction ID

User State PDA

An on-chain profile account for a user.

What gets stored?

Profile info (name, profile picture, bio, etc.)
Number of uploaded files
Friend request records

Note: Friend requests are not stored as values in the PDA; they are sent as transactions.

When is it created?

It is created automatically the first time you call code_in(). No extra setup is required, but the first user may need to sign twice.

Connection PDA

An on-chain account that manages relationships between two users (friends, messages, etc.).

What states can it have?

pending: a friend request was sent but not accepted yet
approved: the request was accepted and the users are connected
blocked: one side blocked the other

Important: A blocked connection can only be unblocked by the blocker.

Key related functions

request_connection(): send a friend request (creates pending)
manage_connection(): approve/reject/block/unblock a request
read_connection(): check current relationship status
write_connection_row(): exchange messages/data with a connected friend
fetch_user_connections(): fetch all connections (sent & received friend requests)

Database Tables

Store JSON data in tables like a database.

How are tables created?

You can create tables explicitly with create_table(), or implicitly — the first write via write_row() creates the table automatically.

Note: A table is uniquely identified by the combination of db_root_id and table_seed (table name).

Key related functions

create_table(): create a table explicitly
write_row(): add a new row (creates the table if missing)
read_table_rows(): read rows from a table
get_tablelist_from_root(): list all tables in a database
fetch_inventory_transactions(): list uploaded files

Token & Collection Gating

Tables can be gated so that only users holding a specific token or NFT collection can write data.

Gate Types

Type	`GateType`	Description
Token	`GateType.TOKEN`	User must hold >= `amount` of the specified SPL token mint
Collection	`GateType.COLLECTION`	User must hold any NFT from the specified Metaplex verified collection

How it works

Table creator sets the gate when creating or updating a table
Writers don't need to do anything special — the SDK automatically resolves the required token account (and metadata account for collections) when calling write_row() or manage_row_data()
If no gate is set, the table is public (default behavior, no change for existing users)

Gate parameter

gate = {
    "mint": Pubkey,       # token mint address OR collection address
    "amount": int,        # minimum token amount (default: 1, ignored for collections)
    "gate_type": int,     # GateType.TOKEN (default) or GateType.COLLECTION
}

Notes

For token gates, amount specifies the minimum balance required (e.g., 100 means "must hold >= 100 tokens")
For collection gates, the user can present any NFT from that collection. amount is ignored (NFTs always have amount=1)
Omitting the gate parameter creates a public table with no restrictions

Encryption (Crypto)

The SDK includes a built-in encryption module (iqlabs.crypto) for encrypting data before storing it on-chain.

Three encryption modes

DH Encryption (single recipient): Ephemeral X25519 ECDH → HKDF-SHA256 → AES-256-GCM. Use when encrypting data for one specific recipient.
Password Encryption: PBKDF2-SHA256 (250k iterations) → AES-256-GCM. Use for password-protected data that anyone with the password can decrypt.
Multi-recipient Encryption (PGP-style hybrid): Generates a random content encryption key (CEK), encrypts data once, then wraps the CEK for each recipient via ECDH. Use when encrypting data for multiple recipients.

Key derivation

Users can derive a deterministic X25519 keypair from their wallet signature using derive_x25519_keypair(). This means users don't need to manage separate encryption keys — their wallet is the key.

Key related functions

derive_x25519_keypair(): derive encryption keypair from wallet
dh_encrypt() / dh_decrypt(): single-recipient encryption
password_encrypt() / password_decrypt(): password-based encryption
multi_encrypt() / multi_decrypt(): multi-recipient encryption

Function Details

Data Storage and Retrieval

`code_in()`

Parameters	`connection`: Solana RPC AsyncClient `signer`: Keypair or WalletSigner `chunks`: data to upload (list[str]) `filename`: optional filename (str or None) `method`: upload method (int, default: 0) `filetype`: file type hint (str, default: '') `on_progress`: optional progress callback (Callable[[int], None])
Returns	Transaction signature (str)

Example:

from iqlabs import writer
from solana.rpc.async_api import AsyncClient
from solders.keypair import Keypair

# Upload data
signature = await writer.code_in(connection, signer, ['Hello, blockchain!'])

# Upload with filename
signature = await writer.code_in(connection, signer, ['file contents here'], filename='hello.txt')

`read_code_in()`

Parameters	`tx_signature`: transaction signature (str) `speed`: rate limit profile (optional, str) `on_progress`: optional progress callback (Callable[[int], None])
Returns	dict with `metadata` (str) and `data` (str or None)

Example:

from iqlabs import reader

result = await reader.read_code_in('5Xg7...')
print(result['data'])      # 'Hello, blockchain!'
print(result['metadata'])  # JSON string with file metadata

Connection Management

`request_connection()`

Parameters	`connection`: AsyncClient `signer`: Keypair or WalletSigner `db_root_id`: database ID (bytes or str) `party_a`: first user pubkey (str) `party_b`: second user pubkey (str) `table_name`: connection table name (str or bytes) `columns`: column list (list[str or bytes]) `id_col`: ID column (str or bytes) `ext_keys`: extension keys (list[str or bytes])
Returns	Transaction signature (str)

Example:

from iqlabs import writer

await writer.request_connection(
    connection, signer, 'my-db',
    my_wallet_address, friend_wallet_address,
    'dm_table', ['message', 'timestamp'], 'message_id', []
)

`manage_connection()`

Note: There is no high-level SDK wrapper for this function. Use the contract-level instruction builder directly.

Parameters	`builder`: InstructionBuilder `accounts`: dict with `db_root`, `connection_table`, `signer` `args`: dict with `db_root_id`, `connection_seed`, `new_status`
Returns	Instruction

Example:

from iqlabs import contract

# Create an instruction builder
builder = contract.create_instruction_builder(contract.PROGRAM_ID)

# Approve a friend request
approve_ix = contract.manage_connection_instruction(
    builder,
    {"db_root": db_root, "connection_table": connection_table, "signer": my_pubkey},
    {"db_root_id": db_root_id, "connection_seed": connection_seed, "new_status": contract.CONNECTION_STATUS_APPROVED}
)

# Block a user
block_ix = contract.manage_connection_instruction(
    builder,
    {"db_root": db_root, "connection_table": connection_table, "signer": my_pubkey},
    {"db_root_id": db_root_id, "connection_seed": connection_seed, "new_status": contract.CONNECTION_STATUS_BLOCKED}
)

`read_connection()`

Parameters	`db_root_id`: database ID (bytes or str) `party_a`: first wallet (str) `party_b`: second wallet (str)
Returns	dict with `status`, `requester`, `blocker`

Example:

from iqlabs import reader

conn_info = await reader.read_connection('my-db', party_a, party_b)
print(conn_info['status'])  # 'pending' | 'approved' | 'blocked'

`write_connection_row()`

Parameters	`connection`: AsyncClient `signer`: Keypair or WalletSigner `db_root_id`: database ID (bytes or str) `connection_seed`: connection seed (bytes or str) `row_json`: JSON data (str)
Returns	Transaction signature (str)

Example:

from iqlabs import writer
import json

await writer.write_connection_row(
    connection, signer, 'my-db', connection_seed,
    json.dumps({"message_id": "123", "message": "Hello friend!", "timestamp": 1234567890})
)

`fetch_user_connections()`

Fetch all connections (friend requests) for a user by analyzing their UserState PDA transaction history. Each connection includes its db_root_id, identifying which app the connection belongs to.

Parameters	`user_pubkey`: user public key (str or Pubkey) `limit`: max number of transactions to fetch (optional) `before`: signature to paginate from (optional) `speed`: rate limit profile (optional)
Returns	List of connection dicts with db_root_id, connection_pda, party_a, party_b, status, requester, blocker, timestamp

Example:

from iqlabs import reader

connections = await reader.fetch_user_connections(
    my_pubkey,
    speed="light",
    limit=100
)

# Filter by status
pending_requests = [c for c in connections if c['status'] == 'pending']
friends = [c for c in connections if c['status'] == 'approved']
blocked = [c for c in connections if c['status'] == 'blocked']

# Check connection details
for conn in connections:
    print(f"Party A: {conn['party_a']} <-> Party B: {conn['party_b']}, status: {conn['status']}")

`update_user_metadata()`

Parameters	`connection`: AsyncClient `signer`: Keypair or WalletSigner `db_root_id`: database ID (bytes or str) `meta`: metadata to store (bytes or str)
Returns	Transaction signature (str)

Example:

from iqlabs import writer

# Store a codeIn tx signature as user metadata
tx_sig = await writer.update_user_metadata(
    connection, signer, 'my-db', meta_tx_signature
)

Table Management

`create_table()`

Parameters connection: AsyncClient
signer: Keypair or WalletSigner
db_root_id: database ID (bytes or str)
table_seed: table seed (bytes or str)
table_name: display name (str or bytes)
column_names: column names (list[str or bytes])
id_col: ID column (str or bytes)
ext_keys: extension keys (list[str or bytes])
gate: optional access gate (see Token & Collection Gating)
writers: optional writer whitelist (list[Pubkey])

Returns Transaction signature (str)

Example:

from iqlabs import writer
from iqlabs.contract import GateType

# No gate (public table)
await writer.create_table(
    connection, signer, 'my-db', 'users', 'Users Table',
    ['name', 'email'], 'user_id', []
)

# With token gate (must hold >= 100 tokens)
await writer.create_table(
    connection, signer, 'my-db', 'vip', 'VIP Table',
    ['name'], 'user_id', [],
    gate={"mint": token_mint_pubkey, "amount": 100, "gate_type": GateType.TOKEN}
)

# With NFT collection gate
await writer.create_table(
    connection, signer, 'my-db', 'holders', 'Holder Table',
    ['name'], 'user_id', [],
    gate={"mint": collection_pubkey, "gate_type": GateType.COLLECTION}
)

`write_row()`

Parameters	`connection`: AsyncClient `signer`: Keypair or WalletSigner `db_root_id`: database ID (bytes or str) `table_seed`: table name (bytes or str) `row_json`: JSON row data (str) `skip_confirmation`: skip tx confirmation (default: False)
Returns	Transaction signature (str)

Example:

from iqlabs import writer
import json

# Write the first row to create the table
await writer.write_row(connection, signer, 'my-db', 'users', json.dumps({
    "id": 1, "name": "Alice", "email": "alice@example.com"
}))

# Add another row to the same table
await writer.write_row(connection, signer, 'my-db', 'users', json.dumps({
    "id": 2, "name": "Bob", "email": "bob@example.com"
}))

`read_table_rows()`

Parameters	`account`: table PDA (Pubkey or str) `before`: signature cursor for pagination (optional) `limit`: max number of rows to fetch (optional) `speed`: rate limit profile (optional)
Returns	`list[dict]`

Example:

from iqlabs import reader

# Basic usage
rows = await reader.read_table_rows(table_pda, limit=50)

# Cursor-based pagination
older_rows = await reader.read_table_rows(table_pda, limit=50, before="sig...")

`get_tablelist_from_root()`

Parameters	`connection`: AsyncClient `db_root_id`: database ID (bytes or str)
Returns	dict with `root_pda`, `creator`, `table_seeds`, `global_table_seeds`

Example:

from iqlabs import reader

result = await reader.get_tablelist_from_root(connection, 'my-db')
print('Creator:', result['creator'])
print('Table seeds:', result['table_seeds'])

`fetch_inventory_transactions()`

Parameters	`public_key`: user public key (Pubkey) `limit`: max count (int) `before`: pagination cursor (optional, str)
Returns	Transaction list

Example:

from iqlabs import reader
import json

my_files = await reader.fetch_inventory_transactions(my_pubkey, 20)
for tx in my_files:
    metadata = None
    try:
        metadata = json.loads(tx['metadata'])
    except:
        metadata = None

    if metadata and 'data' in metadata:
        inline_data = metadata['data'] if isinstance(metadata['data'], str) else json.dumps(metadata['data'])
        print(f"Inline data: {inline_data}")
    else:
        print(f"Signature: {tx['signature']}")

Encryption

`derive_x25519_keypair()`

Derive a deterministic X25519 keypair from a wallet signature. The same wallet always produces the same keypair.

Parameters	`sign_message`: async sign function `Callable[[bytes], Awaitable[bytes]]`
Returns	dict with `priv_key` (bytes) and `pub_key` (bytes)

Example:

from iqlabs import crypto

keypair = await crypto.derive_x25519_keypair(wallet.sign_message)
pub_hex = keypair['pub_key'].hex()

`dh_encrypt()`

Parameters	`recipient_pub_hex`: recipient's X25519 public key (hex str) `plaintext`: data to encrypt (bytes)
Returns	dict with `sender_pub`, `iv`, `ciphertext` (all hex str)

`dh_decrypt()`

Parameters	`priv_key`: recipient's private key (bytes) `sender_pub_hex`: sender's public key from encrypt result (hex str) `iv_hex`: IV from encrypt result (hex str) `ciphertext_hex`: ciphertext from encrypt result (hex str)
Returns	bytes (decrypted plaintext)

Example:

from iqlabs import crypto

# Encrypt
encrypted = crypto.dh_encrypt(recipient_pub_hex, b'secret message')

# Decrypt (recipient side)
decrypted = crypto.dh_decrypt(
    recipient_priv_key, encrypted['sender_pub'], encrypted['iv'], encrypted['ciphertext']
)

`password_encrypt()`

Parameters	`password`: password (str) `plaintext`: data to encrypt (bytes)
Returns	dict with `salt`, `iv`, `ciphertext` (all hex str)

`password_decrypt()`

Parameters	`password`: password (str) `salt_hex`: salt from encrypt result (hex str) `iv_hex`: IV from encrypt result (hex str) `ciphertext_hex`: ciphertext from encrypt result (hex str)
Returns	bytes (decrypted plaintext)

Example:

from iqlabs import crypto

encrypted = crypto.password_encrypt('my-password', b'secret data')
decrypted = crypto.password_decrypt(
    'my-password', encrypted['salt'], encrypted['iv'], encrypted['ciphertext']
)

`multi_encrypt()`

Parameters	`recipient_pub_hexes`: recipient public keys (list[str]) `plaintext`: data to encrypt (bytes)
Returns	dict with `recipients` (list[RecipientEntry]), `iv`, `ciphertext`

`multi_decrypt()`

Parameters	`priv_key`: your private key (bytes) `pub_key_hex`: your public key (hex str) `encrypted`: the MultiEncryptResult dict
Returns	bytes (decrypted plaintext)

Example:

from iqlabs import crypto

# Encrypt for multiple recipients
encrypted = crypto.multi_encrypt(
    [alice_pub_hex, bob_pub_hex, carol_pub_hex],
    b'group secret'
)

# Each recipient decrypts with their own key
plaintext = crypto.multi_decrypt(alice_priv_key, alice_pub_hex, encrypted)

Environment Settings

`set_rpc_url()`

Parameters	`url`: Solana RPC URL (str)
Returns	None

Example:

from iqlabs import set_rpc_url

set_rpc_url('https://your-rpc.example.com')

Helius Optimization

When the RPC URL points to a Helius endpoint (helius-rpc.com), the SDK automatically uses getTransactionsForAddress (gTFA) for reading session files. This fetches 100 full transactions per call instead of individual getTransaction calls.

Result: Large file reads are ~100x faster (e.g. 580KB file: 468s → 4.6s).

No code changes needed — just set a Helius RPC URL:

from iqlabs import set_rpc_url

set_rpc_url('https://mainnet.helius-rpc.com/?api-key=YOUR_KEY')

# read_code_in automatically uses gTFA when available
result = await reader.read_code_in(tx_signature)

Falls back to standard sequential reads on any non-Helius RPC. Requires a paid Helius plan for gTFA access.

Advanced Functions

These functions are advanced/internal, so this doc lists them only. For details, please see our developer docs.

manage_row_data() (writer)
read_user_state() (reader)
read_inventory_metadata() (reader)
read_user_inventory_code_in_from_tx() (reader)
get_session_pda_list() (reader)
derive_dm_seed() (utils)
to_seed_bytes() (utils)
hex_to_bytes() / bytes_to_hex() / validate_pub_key() (crypto)

Name		Name	Last commit message	Last commit date
Latest commit History 20 Commits
.idea		.idea
examples		examples
idl		idl
iqlabs		iqlabs
iqlabs_solana_sdk.egg-info		iqlabs_solana_sdk.egg-info
tests		tests
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
README.md		README.md
main.py		main.py
plan.md		plan.md
pyproject.toml		pyproject.toml

Folders and files

Latest commit

History

Repository files navigation

IQLabs SDK (Python)

Table of Contents

Core Concepts

Data Storage (Code In)

How is it stored?

Key related functions

User State PDA

What gets stored?

When is it created?

Connection PDA

What states can it have?

Key related functions

Database Tables

How are tables created?

Key related functions

Token & Collection Gating

Gate Types

How it works

Gate parameter

Notes

Encryption (Crypto)

Three encryption modes

Key derivation

Key related functions

Function Details

Data Storage and Retrieval

code_in()

read_code_in()

Connection Management

request_connection()

manage_connection()

read_connection()

write_connection_row()

fetch_user_connections()

update_user_metadata()

Table Management

create_table()

write_row()

read_table_rows()

get_tablelist_from_root()

fetch_inventory_transactions()

Encryption

derive_x25519_keypair()

dh_encrypt()

dh_decrypt()

password_encrypt()

password_decrypt()

multi_encrypt()

multi_decrypt()

Environment Settings

set_rpc_url()

Helius Optimization

Advanced Functions

Additional Resources

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

`code_in()`

`read_code_in()`

`request_connection()`

`manage_connection()`

`read_connection()`

`write_connection_row()`

`fetch_user_connections()`

`update_user_metadata()`

`create_table()`

`write_row()`

`read_table_rows()`

`get_tablelist_from_root()`

`fetch_inventory_transactions()`

`derive_x25519_keypair()`

`dh_encrypt()`

`dh_decrypt()`

`password_encrypt()`

`password_decrypt()`

`multi_encrypt()`

`multi_decrypt()`

`set_rpc_url()`

Packages