Developer Docs (Python)

This document is in progress and will be refined.

Installation

pip install iqlabs-solana-sdk

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

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

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

A blocked connection can only be unblocked by the blocker.

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?

There is no dedicated “create table” function. The first write via write_row() creates the table automatically.

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

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

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)

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)

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

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()`

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

from iqlabs import contract

# Create an instruction builder
builder = contract.create_instruction_builder(contract.get_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`

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 `db_root_id`: database ID (bytes or str) `connection_seed`: connection seed (bytes or str) `row_json`: JSON data (str)
Returns	Transaction signature (str)

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

from iqlabs import reader

# Fetch all connections (across all apps!)
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']}")

Table Management

`write_row()`

Parameters	`connection`: AsyncClient `signer`: Keypair `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)

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]`

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`

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

from iqlabs import reader
from solders.pubkey import Pubkey
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']}")

Environment Settings

`set_rpc_url()`

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

from iqlabs import set_rpc_url

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

Advanced Functions

These are low-level SDK functions. Not needed for typical usage, but useful when building custom features or debugging.

Writer Functions

`manage_row_data()`

Unified function that handles both table row writes and connection row writes. Auto-detects whether to write to a table or connection based on existing PDAs.

Module	`writer`
Parameters	`connection`: AsyncClient `signer`: Keypair `db_root_id`: database ID (bytes or str) `seed`: table or connection seed (bytes or str) `row_json`: JSON row data (str) `table_name`: required for table edits (optional) `target_tx`: reference tx for table edits (optional)
Returns	Transaction signature (str)
Use Case	Custom row management, updating existing rows

from iqlabs import writer
import json

await writer.manage_row_data(
    connection, signer,
    'my-db',       # db_root_id
    'users',       # seed (table seed)
    json.dumps({"id": 1, "name": "Updated Name"}),
    table_name='users',
    target_tx=original_tx_sig
)

Reader Functions

`read_user_state()`

Reads the UserState PDA for a given user.

Module	`reader`
Parameters	`user_pubkey`: user public key (str)
Returns	dict with `owner`, `metadata`, `total_session_files`, `profile_data`
Use Case	Fetching user profile data, checking upload counts

from iqlabs import reader

user_state = await reader.read_user_state(user_pubkey)
print('Owner:', user_state['owner'])
print('Session files:', user_state['total_session_files'])
print('Profile data:', user_state['profile_data'])

`read_inventory_metadata()`

Reads metadata associated with a user’s inventory transaction.

Module	`reader`
Parameters	`tx_signature`: transaction signature (str)
Returns	dict with metadata
Use Case	Extracting file metadata from inventory transactions

from iqlabs import reader

result = await reader.read_inventory_metadata(tx_signature)
print('Metadata:', result)

`get_session_pda_list()`

Retrieves a list of session PDA addresses for a user.

Module	`reader`
Parameters	`user_pubkey`: user public key (str)
Returns	`list[str]` (session PDA base58 strings)
Use Case	Session management, active session tracking

from iqlabs import reader

sessions = await reader.get_session_pda_list(user_pubkey)
for pda in sessions:
    print(f"Session PDA: {pda}")

Contract PDA Functions

Low-level PDA derivation functions available in the contract module.

Module	`contract`
Use Case	Custom PDA derivation, account lookups

from iqlabs import contract
from solders.pubkey import Pubkey

program_id = contract.get_program_id()
db_root_pda = contract.get_db_root_pda(db_root_id, program_id)
table_pda = contract.get_table_pda(db_root_pda, table_seed, program_id)
user_pda = contract.get_user_pda(user_pubkey, program_id)
session_pda = contract.get_session_pda(user_pubkey, seq=0, program_id=program_id)
connection_pda = contract.get_connection_table_pda(db_root_pda, connection_seed, program_id)

Utility Functions

`derive_dm_seed()`

Derives a deterministic seed for direct messaging (DM) between two users. Sorts the two pubkeys alphabetically and hashes "lower:upper" with Keccak-256.

Module	`utils`
Parameters	`user_a`: first user pubkey (str) `user_b`: second user pubkey (str)
Returns	bytes (32-byte seed)
Use Case	Creating consistent connection identifiers, DM channel setup

from iqlabs.sdk.utils.seed import derive_dm_seed

seed1 = derive_dm_seed(wallet_a, wallet_b)
seed2 = derive_dm_seed(wallet_b, wallet_a)
print(seed1 == seed2)  # True — order doesn't matter

`to_seed_bytes()`

Converts a seed identifier to bytes. If the input is a 64-character hex string, it passes through as-is. Otherwise, it applies Keccak-256 hash.

Module	`utils`
Parameters	`value`: seed string or bytes
Returns	bytes
Use Case	Custom PDA derivation, low-level seed manipulation

from iqlabs.sdk.utils.seed import to_seed_bytes
from solders.pubkey import Pubkey

seed_string = 'my-custom-seed'
seed_bytes = to_seed_bytes(seed_string)

pda, bump = Pubkey.find_program_address(
    [seed_bytes, other_seed],
    program_id
)

Page

​Installation

​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

​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()

​Table Management

​write_row()

​read_table_rows()

​get_tablelist_from_root()

​fetch_inventory_transactions()

​Environment Settings

​set_rpc_url()

​Advanced Functions

​Writer Functions

​manage_row_data()

​Reader Functions

​read_user_state()

​read_inventory_metadata()

​get_session_pda_list()

​Contract PDA Functions

​Utility Functions

​derive_dm_seed()

​to_seed_bytes()

Installation

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

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()`

Table Management

`write_row()`

`read_table_rows()`

`get_tablelist_from_root()`

`fetch_inventory_transactions()`

Environment Settings

`set_rpc_url()`

Advanced Functions

Writer Functions

`manage_row_data()`

Reader Functions

`read_user_state()`

`read_inventory_metadata()`

`get_session_pda_list()`

Contract PDA Functions

Utility Functions

`derive_dm_seed()`

`to_seed_bytes()`