Python SDK

El SDK de Python d’Asset Core proporciona un client HTTP tipat per interactuar amb els daemons d’escriptura i lectura, a més de constructors d’operacions que es mapegen directament al JSON de transacció. És un envoltori lleuger sobre el contracte HTTP públic, així que qualsevol cosa que puguis fer en el SDK també es pot fer a través de HTTP brut.

Visió general

El SDK es distribueix com el paquet assetcore-sdk i proporciona:

• AssetCoreClient: Client d’alt nivell amb mètodes asíncrons i sincrons
• Constructor d’operacions: Classes de dades tipades per a les 24 operacions
• Tipus d’error: Excepcions estructurades per al maneig d’errors
• Client generat: capa HTTP basada en OpenAPI

El SDK és un client pur; invoca l’API HTTP pública i no pot mutar l’estat del servidor directament. Això manté el model de seguretat consistent a través de totes les integracions.

Estructura

assetcore_sdk/
  __init__.py          # Package exports
  client.py            # AssetCoreClient facade
  operations.py        # Operation builder dataclasses
  errors.py            # Exception types
  models/              # Typed data classes
  utils/               # HTTP and serialization helpers
  _generated/          # OpenAPI codegen output

Campanyes

AssetCoreClient

La classe principal del client per interactuar amb Asset Core. Utilitzeu instàncies de client separades per a tokens d’escriptura i lectura per preservar l’accés amb el mínim privilegi.

import os
from assetcore_sdk import AssetCoreClient

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
    timeout=30.0,
)

read_client = AssetCoreClient(
    base_url="http://localhost:8081",
    api_key=os.environ["ASSETCORE_READ_TOKEN"],
    timeout=30.0,
)

Paràmetre	Tipus	Per defecte	Descripció
base_url	str	requerit	URL base per al daemon d’escriptura o lectura
api_key	str	requerit	Token Bearer utilitzat per a Authorization
timeout	float	10.0	Temps d’espera de la sol·licitud en segons
headers	dict	Cap	Encapsulaments addicionals a enviar amb cada sol·licitud
refresh_token	callable	Cap	Proveïdor opcional de token de refresc de sincronització
refresh_token_async	callable	Cap	Proveïdor opcional de token de refresc asíncron

Mètodes de Compromís

# Async
result = await write_client.commit_operations(
    operations=[...],
    idempotency_key="unique-key-2026-01-15",
    namespace_id=5001
)

# Sync
result = write_client.commit_operations_sync(
    operations=[...],
    idempotency_key="unique-key-2026-01-15",
    namespace_id=5001
)

Paràmetre	Tipus	Requerit	Descripció
operations	llista	Sí	Llista d’operacions tipades o sobres
namespace_id	int	No	Espai de noms (per defecte és 1)
idempotency_key	str	No	Clau de deduplicació
metadata	dict	No	Metadades de l’usuari
actor_id	str	No	Identificador de l’actor per a les pistes d’auditoria
policy_id	str	No	Identificador de la política (ús futur)
origin	dict	No	Metadades d’origen opcionals

Retorna CommitResponse amb world_seq_start, world_seq_end, event_count i metadades de correlació.

Llegir Mètodes

# Get container balances
balances = await read_client.get_container_balances(1001)

# Get container (sync)
container = read_client.get_container_sync(1001)

Els mètodes de lectura retornen objectes de resposta tipats amb metadades de frescor. Podeu utilitzar la frescor per fer complir la semàntica de llegir les vostres escriptures en el codi del client.

Constructors d’Operacions

Importa constructors individuals o tot el mòdul:

from assetcore_sdk.operations import (
    CreateContainer,
    AddFungible,
    RemoveFungible,
    TransferFungible,
    AddInstance,
    PlaceInSlot,
    RegisterClass,
)

Cada constructor crea una dataclass que es pot convertir a l’envolup d’operació:

op = CreateContainer(
    container_id=1001,
    kind={"type": "balance"},
    owner=None,
    policies=None,
)
# op.to_envelope() -> {"op": "CreateContainer", "args": {...}}

Tipus d’Error

from assetcore_sdk.errors import (
    AssetCoreError,        # Base exception
    HttpError,             # Transport failures
    ValidationError,       # 422 errors
    ConflictError,         # 409 errors
    NotFoundError,         # 404 errors
    SDKUnauthorizedError,  # 401 errors
    SDKForbiddenError,     # 403 errors
    SDKQuotaError,         # 429 errors
    TokenRefreshError      # Refresh hook failures
)

Els errors inclouen el cos de resposta amb el codi d’error i detalls.

Exemples

Compromís Bàsic

from assetcore_sdk import AssetCoreClient
from assetcore_sdk.operations import CreateContainer, AddFungible

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
)

operations = [
    CreateContainer(
        container_id=1001,
        kind={"type": "balance"},
        owner=None,
        policies=None,
    ),
    AddFungible(
        class_id=100,
        key=1,
        quantity=500,
        location={"container_id": 1001, "kind": "balance"},
    ),
]

result = await write_client.commit_operations(
    operations,
    idempotency_key="init-container-2026-01-15-1001",
    namespace_id=5001,
)

print(f"Committed at world seq {result.world_seq_start}")

Gestió d’Errors

from assetcore_sdk.errors import ValidationError, ConflictError

try:
    result = await write_client.commit_operations(operations, namespace_id=5001)
except ValidationError as e:
    print(f"Validation failed: {e.error_code}")
    print(f"Details: {e.details}")
except ConflictError as e:
    print(f"Conflict: {e.message}")

Estat de lectura

# Get balances
balances = await read_client.get_container_balances(1001, namespace_id=5001)
for balance in balances.balances:
    print(f"Class {balance.class_id}: {balance.quantity}")

# Check freshness
print(f"Data as of world seq {balances.freshness.world_seq}")

Ús de la sincronització

Per a scripts o entorns sense async:

import os
from assetcore_sdk import AssetCoreClient

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
)

read_client = AssetCoreClient(
    base_url="http://localhost:8081",
    api_key=os.environ["ASSETCORE_READ_TOKEN"],
)

# All async methods have _sync variants
result = write_client.commit_operations_sync(operations, namespace_id=5001)
balances = read_client.get_container_balances_sync(1001, namespace_id=5001)

Construcció de Transaccions Complexes

from assetcore_sdk.operations import (
    RegisterClass,
    RegisterClassShape,
    AddInstance,
)

operations = [
    RegisterClass(
        request={
            "class_id": 200,
            "flags": 2,
            "name": "Sample Tube",
        }
    ),
    RegisterClassShape(
        request={
            "class_id": 200,
            "stack_key": None,
            "shape": {"width": 1, "height": 3},
        }
    ),
    AddInstance(
        class_id=200,
        key=1,
        location={"container_id": 1001, "kind": "slot", "slot_index": 1},
    ),
]

result = await write_client.commit_operations(operations, namespace_id=5001)
print(f"Committed transaction {result.commit_id}")

Referències relacionades

• Introducció al SDK de Python
• Referència d’Acció
• Receptes