sync_kubo — Network¶

Version: 19.0.0.3.0 · Depends: sync_base · License: OPL-1

sync_kubo connects Odoo to a running Kubo daemon via a Python sidecar gateway. It is the natural choice for large binary payloads, offline-first networks, and deployments that already run an IPFS node.

Architecture¶

Odoo (sync_kubo)
     │  JSON-RPC
     ▼
Python Gateway (gateway/)
     │  Kubo RPC         │  Express API
     ▼                   ▼
Kubo daemon          /express-send
/api/v0/*            /express-fetch

The gateway bridges Odoo JSON-RPC calls with the Kubo daemon’s /api/v0/ HTTP API, and exposes an express API for high-priority operations.

Models¶

class ip.swarm[source]¶

Kubo transport implementation of the network dispatch hook.

Implements the dispatch and state-reconciliation hooks for Kubo transport.

Outbound Dispatch

_dispatch_events(events)[source]¶

Batch all events into a single express_send POST per swarm gateway (fire-and-forget).

Delegates to ip.node._kubo_send_events so the transport logic lives in one place. Events remain in ‘sending’ state; the Postman cron retries any the gateway drops.

CID Fetching

_do_express_fetch_cids(cids)[source]¶: POST wanted CIDs to the Kubo gateway’s express-fetch queue (fire-and-forget).

Gateway Sync

_safe_execute_batch(batch_name: str, func: Callable, *args) → dict[source]¶

Safely execute a batch operation within a savepoint to isolate failures.

Protects critical operations without sacrificing bulk performance.

Parameters:

batch_name – Name of the batch (for logging).
func – Callable to execute.
args – Additional positional arguments forwarded to func.

Returns:

{"info": <func return value>} on success, {"error": <exception message>} on soft failure. PostgreSQL serialisation/deadlock errors are re-raised so Odoo’s HTTP dispatcher can retry the transaction automatically.

_sync_get_push_order()[source]¶: Return the model processing order for inbound push batches. CIDs must precede Events.

_sync_get_update_order()[source]¶: Return the model processing order for inbound update batches.

_sync_get_pull_order()[source]¶: Return the model processing order for outbound pull responses.

_sync_gateway_state(sync_payload: dict, last_sync: str | datetime) → dict[source]¶

Execute a full two-way state reconciliation for the swarm’s gateway.

Processes pushes and updates from the gateway, then prepares a pull response with current Odoo state. All pushes run inside savepoints via _safe_execute_batch so a single failing model never rolls back others.

Parameters:

sync_payload – Grouped actions from the gateway, e.g. {"push": {"ip.event": [...]}, "update": {...}}.
last_sync – Watermark timestamp of the previous successful sync (used to filter stale pull results). False to fetch all.

Returns:

Reconciled state dict with keys push, update, pull, and last_sync (the new safe watermark for the next cycle).

class ip.node[source]¶

Kubo / Libp2p transport extension for ip.node.

Extends the base node with Kubo daemon and gateway endpoints.

Attributes

kubo_url: fields.Char[source]¶: Kubo/Sync sidecar API endpoint, used for direct CID management (add / cat).

gate_url: fields.Char[source]¶: Gateway API endpoint, used for express requests (send events / fetch CIDs).

Transport Interface

_kubo_send_events(payloads: list) → bool[source]¶: Batch-deliver payloads to the Kubo gateway’s express-send endpoint.

add(payload)[source]¶

Add content to Sync and return the pinned ip.cid record.

Parameters:: payload – The data to upload (str, bytes, or dict). Dicts are serialised to JSON before upload.
Returns:: The ip.cid record (state=pinned), or None if the node is unreachable or the request fails.

cat(cid: str | models.BaseModel) → bytes | None[source]¶

Fetch content by CID (IPFS CIDv1).

Checks data_raw in the local filestore first; so this avoids a Kubo RPC call in the best case. Falls back to calling Kubo’s /api/v0/cat endpoint on a cache miss.

Parameters:: cid – CID string or ip.cid record.
Returns:: Raw bytes of the content, or None if not found / unreachable.

Kubo Helpers

_normalize_payload(payload)[source]¶: Ensure payload is in a format suitable for Kubo API. Auto-serialize dicts to JSON.

_kubo_request(endpoint, params=None, files=None, json=None, timeout=5.0)[source]¶

POST to the Kubo RPC API at kubo_url + endpoint.

Returns:: Response object on success, None on network error or missing URL. Callers are responsible for parsing resp.content / resp.json().

_gate_request(endpoint, json=None, timeout=0.1) → bool[source]¶

Fire-and-forget POST to the Kubo gateway at gate_url + endpoint.

A read-timeout is treated as success (request was dispatched, response not awaited). Returns True if sent or timed-out, False only when the gateway is completely unreachable.

Connectivity

_fetch_identity()[source]¶

Query the Kubo RPC /api/v0/id endpoint and write the node’s cryptographic identity back to the database.

On success, peer_char and public_key on the ip.node record are updated with the values returned by Kubo. Returns the subset of nodes successfully reached; logs warnings for failures instead of raising, to support the polymorphic loopback-probe pattern.

Gateway Sync

_sync_push(vals_list)[source]¶: Push using peer_char as the sync key.

_sync_pull(last_sync)[source]¶: Pull the modified nodes.

class ip.topic[source]¶

Kubo gateway sync extension for ip.topic.

Overrides _sync_pull() to scope the pulled topics to the calling gateway user’s swarm, preventing cross-swarm topic leakage.

Methods

_sync_pull(last_sync)[source]¶: Pull the modified topics, scoped to the calling gateway user’s swarm.