Python

Requiere Python 3.10 o posterior. Está construido sobre httpx y Pydantic v2, e incluye anotaciones de tipos completas que pasan mypy y pyright en modo estricto.

Instalación

pip install stockshift

Inicio rápido

Crea un artículo y luego lista las existencias disponibles:

import os

from stockshift import StockShift

with StockShift(token=os.environ["STOCKSHIFT_TOKEN"]) as stockshift:
    stockshift.catalog.create_item(
        sku="BOLT-M4", name="M4 hex bolt", type="stocked", make_or_buy="buy"
    )

    for row in stockshift.inventory.iter_stock_on_hand():
        print(row.sku, row.location_code, row.available_to_promise)

AsyncStockShift tiene los mismos métodos para asyncio y Trio. Usa await en cada llamada y pagina con async for:

from stockshift import AsyncStockShift

async with AsyncStockShift(token=os.environ["STOCKSHIFT_TOKEN"]) as stockshift:
    async for item in stockshift.catalog.iter_items(search="bolt"):
        print(item.sku)

Autenticación

Pasa una clave de API o un token de acceso OAuth como token. Crea una clave de API en la app, en Configuración > Acceso > Claves de API. La clave actúa como el miembro que la creó, y cada llamada necesita tanto el permiso del miembro como uno de los alcances de la clave. Para un token OAuth que vence, pasa una función. El cliente la llama antes de cada solicitud. AsyncStockShift también acepta una función de corrutina:

stockshift = StockShift(token=lambda: tokens.current())

Otros argumentos:

Argumento	Valor predeterminado	Significado
`base_url`	`https://app.stockshift.io/api/v1`	Dónde está la API.
`max_retries`	`2`	Reintentos para una llamada que se puede reintentar.
`timeout`	`60`	Segundos de espera para cada intento.
`max_retry_after`	`60`	El `Retry-After` más largo, en segundos, que vale la pena esperar.
`headers`	ninguno	Encabezados que se envían en cada llamada.
`http_client`	uno nuevo	Un `httpx.Client` o `httpx.AsyncClient` con el que enviar.

with_options() devuelve una copia con otros ajustes que comparte las conexiones:

stockshift.with_options(timeout=120, max_retries=5).catalog.list_units_of_measure()

Hacer llamadas

Cada área de la API es un atributo, como catalog o inventory, y cada operación es un método en él. Primero van los parámetros de ruta. Todo lo demás es un argumento con nombre:

item = stockshift.catalog.get_item("01J8QZ4M...")
print(item.sku)

page = stockshift.catalog.list_items(search="bolt", status="active")

El docstring de cada método indica su ruta, como GET /catalog/items/{item}, y copia la referencia de la API. Una llamada devuelve un modelo de Pydantic de stockshift.models, o una lista de ellos. Una respuesta vacía devuelve None, una exportación CSV su texto y un archivo el httpx.Response. Un campo que la API agregue después de esta versión se conserva en el modelo, y puedes leerlo desde model_extra. Un campo opcional del cuerpo que omites no se envía. Pasa None para enviar null, que vacía el campo. Para llamar a una ruta para la que esta versión no tiene método, usa request(). Devuelve el httpx.Response:

response = stockshift.request("GET", "/catalog/items/{item}", path_params={"item": item.id})

Paginación

Una lista que puede crecer devuelve una página a la vez. Cada lista paginada tiene un segundo método que empieza con iter_. Sigue los cursores y entrega cada registro:

for item in stockshift.catalog.iter_items(search="bolt"):
    print(item.sku)

Obtiene la página siguiente solo cuando se agotan los registros de la anterior. Para paginar a mano, pasa el next_cursor de cada página como after hasta que sea None.

Errores

Una llamada fallida lanza un APIError que contiene el cuerpo de detalles del problema de RFC 9457. Decide según code, que es estable. detail está escrito para personas y puede cambiar.

from stockshift import APIError

try:
    stockshift.catalog.create_item(
        sku="BOLT-M4", name="M4 hex bolt", type="stocked", make_or_buy="buy"
    )
except APIError as error:
    if error.code == "validation_failed":
        print(error.errors)  # {'sku': ['Another item already has the SKU BOLT-M4.']}

APIError también tiene status, problem, headers y response. Una llamada que no recibe respuesta, porque falló la red o el intento agotó el tiempo de espera, lanza un APIConnectionError. Ambos extienden StockShiftError, que también se lanza cuando llega una respuesta que esta versión no puede leer.

Reintentos y claves de idempotencia

El cliente reintenta un 429 después del tiempo que indica Retry-After. Lo hace con todos los métodos, porque la API no ejecutó la llamada. Si Retry-After es mayor que max_retry_after, lanza una excepción en lugar de esperar. Un 5xx o una falla de red se reintenta para GET, PUT y DELETE, con una breve espera entre intentos. Un POST o PATCH solo se reintenta cuando le das una clave de idempotencia, para que un reintento no pueda hacer el cambio dos veces:

import uuid

stockshift.integrations.push_stock_now(connection_id, idempotency_key=str(uuid.uuid4()))

Usa una clave nueva para cada cambio que quieras hacer. Los endpoints que respetan la clave lo indican en la referencia de la API.

Webhooks

Revisa el encabezado StockShift-Signature antes de confiar en una entrega. verify_webhook() necesita el cuerpo sin procesar, exactamente como llegó:

from stockshift import WebhookVerificationError, verify_webhook
from stockshift.models import InventoryStockAdjustedEvent


def receive(body: bytes, signature: str | None) -> int:
    try:
        event = verify_webhook(body, signature, os.environ["STOCKSHIFT_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return 400

    if isinstance(event, InventoryStockAdjustedEvent):
        print(event.data.item_id, event.data.quantity)
    return 204

Devuelve un modelo para cada tipo de evento, o un UnknownWebhookEvent para un tipo agregado después de esta versión. Rechaza una entrega de más de cinco minutos de antigüedad. Pasa tolerance, en segundos, para cambiarlo. Los reintentos de un evento conservan su id, así que guárdalo y omite los eventos que ya procesaste. Para probar tu propio receptor, sign_webhook() firma un payload igual que StockShift.

Versiones

La versión mayor cambia cuando la API cambia de una forma que puede romper tu código, como un campo eliminado o un parámetro obligatorio nuevo. Los endpoints y campos nuevos llegan en versiones menores. stockshift.API_VERSION contiene la versión de la referencia de la API con la que se compiló el paquete.

SDK

Frameworks

Instalación

Inicio rápido

Autenticación

Hacer llamadas

Paginación

Errores

Reintentos y claves de idempotencia

Webhooks

Versiones

​Instalación

​Inicio rápido

​Autenticación

​Hacer llamadas

​Paginación

​Errores

​Reintentos y claves de idempotencia

​Webhooks

​Versiones

Instalación

Inicio rápido

Autenticación

Hacer llamadas

Paginación

Errores

Reintentos y claves de idempotencia

Webhooks

Versiones