SDK de Python

El SDK de Python de Asset Core proporciona un cliente HTTP tipado para interactuar con los demonios de escritura y lectura, además de constructores de operaciones que se mapean directamente a JSON de transacción. Es un envoltorio delgado sobre el contrato HTTP público, por lo que cualquier cosa que se pueda hacer en el SDK también se puede hacer a través de HTTP en bruto.

Resumen

El SDK se distribuye como el paquete assetcore-sdk y proporciona:

AssetCoreClient: Cliente de alto nivel con métodos asíncronos y sincrónicos
Constructores de operaciones: Clases de datos tipadas para las 24 operaciones
Tipos de error: Excepciones estructuradas para el manejo de errores
Cliente generado: capa HTTP basada en OpenAPI

El SDK es un cliente puro; invoca la API HTTP pública y no puede mutar el estado del lado del servidor directamente. Esto mantiene el modelo de seguridad consistente en todas las integraciones.

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

Campos

AssetCoreClient

La clase principal del cliente para interactuar con Asset Core. Utilice instancias de cliente separadas para tokens de escritura y lectura para preservar el acceso con el menor privilegio posible.

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ámetro	Tipo	Predeterminado	Descripción
`base_url`	str	requerido	URL base para el daemon de escritura o lectura
`api_key`	str	requerido	Token Bearer utilizado para `Authorization`
`timeout`	float	10.0	Tiempo de espera de la solicitud en segundos
`headers`	dict	Ninguno	Encabezados adicionales para enviar con cada solicitud
`refresh_token`	callable	Ninguno	Proveedor opcional de token de actualización sincrónico
`refresh_token_async`	callable	Ninguno	Proveedor opcional de token de actualización asincrónico

Métodos de Compromiso

# 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ámetro	Tipo	Requerido	Descripción
`operations`	lista	Sí	Lista de operaciones tipadas o sobres
`namespace_id`	int	No	Espacio de nombres (por defecto 1)
`idempotency_key`	str	No	Clave de deduplicación
`metadata`	dict	No	Metadatos del usuario
`actor_id`	str	No	Identificador del actor para auditorías
`policy_id`	str	No	Identificador de la política (uso futuro)
`origin`	dict	No	Metadatos de origen opcionales

Devuelve CommitResponse con world_seq_start, world_seq_end, event_count y metadatos de correlación.

Métodos de Lectura

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

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

Los métodos de lectura devuelven objetos de respuesta tipados con metadatos de frescura. Puede utilizar la frescura para hacer cumplir la semántica de leer-sus-escrituras en el código del cliente.

Constructores de Operaciones

Importar constructores individuales o el módulo completo:

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

Cada constructor crea una clase de datos que se puede convertir en el sobre de operación:

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

Tipos de 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
)

Los errores incluyen el cuerpo de la respuesta con el código de error y detalles.

Ejemplos

Compromiso Básico

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}")

Manejo de Errores

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}")

Estado 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}")

Uso de Sincronización

Para scripts o entornos sin 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)

Construyendo Transacciones Complejas

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}")