Inicio Rápido e Instalación

Guía para instalar el SDK de Python para integración con el BPM (Shori), configurar el cliente y realizar una primera llamada a la API.

---

Requisitos

Antes de comenzar, asegúrate de tener instalado:

• Python 3.10 o superior.
• pip.
• Acceso a un portal de Shori.
• Portal ID, Tenant ID y API Key.

---

Entorno Virtual

Se recomienda utilizar un entorno virtual para aislar las dependencias del proyecto.

Crear el entorno virtual

python -m venv .venv

Activar el entorno

Linux / macOS:

source .venv/bin/activate

Windows / PowerShell:

.\.venv\Scripts\Activate.ps1

Una vez activado, el terminal mostrará (.venv) al inicio de la línea:

(.venv) PS C:\ruta\de\tu\proyecto>

Puedes comprobar que estás utilizando el Python del entorno virtual:

python -c "import sys; print(sys.executable)"

La ruta debe apuntar a:

...\tu-proyecto\.venv\Scripts\python.exe

Después instala el SDK dentro del entorno virtual:

python -m pip install shori-sdk

Para verificar que pip también pertenece al entorno virtual:

python -m pip --version

La ruta mostrada debe contener:

...\tu-proyecto\.venv\Lib\site-packages

Nota para Windows: se recomienda utilizar python -m pip en lugar de pip directamente. De esta forma se garantiza que pip corresponde al mismo intérprete de Python activo.

---

Instalación

Instala el SDK directamente con pip:

python -m pip install shori-sdk

Para instalar una versión específica:

python -m pip install shori-sdk==1.0.2

Puedes verificar la instalación con:

python -m pip show shori-sdk

---

Verificación de la Instalación

Puedes comprobar que el SDK quedó instalado correctamente ejecutando:

python -c "import shori_sdk; print('Shori SDK instalado correctamente')"

También puedes verificar que la clase principal está disponible:

python -c "from shori_sdk import ShoriClientBuilder; print(ShoriClientBuilder)"

Si ambos comandos se ejecutan sin errores, el SDK está disponible en el entorno Python activo.

---

Configuración del Cliente

El cliente se configura utilizando las credenciales de acceso de tu portal:

from shori_sdk import ShoriClientBuilder

client = (
    ShoriClientBuilder()
    .portal_id("tu-portal-id")
    .tenant_id("tu-tenant-id")
    .api_key("tu-api-key")
    .build()
)

Entorno

Cuando sea necesario trabajar con un entorno específico, puedes configurarlo mediante ShoriEnvironment:

from shori_sdk import ShoriClientBuilder, ShoriEnvironment

client = (
    ShoriClientBuilder()
    .portal_id("tu-portal-id")
    .tenant_id("tu-tenant-id")
    .api_key("tu-api-key")
    .environment(ShoriEnvironment.UAT)
    .build()
)

Los entornos disponibles son:

Entorno	Descripción
DEV	Desarrollo
UAT	Pruebas / QA
PROD	Producción

---

Primera Operación

Una vez creado el cliente, puedes utilizar los módulos del SDK.

Por ejemplo, para crear un caso:

from shori_sdk import ShoriClientBuilder

client = (
    ShoriClientBuilder()
    .portal_id("tu-portal-id")
    .tenant_id("tu-tenant-id")
    .api_key("tu-api-key")
    .build()
)

caso = client.caso().create(
    caso_type_id="tu-caso-type-id",
    submitted_data={
        "cliente": "Ana García"
    }
)

print(f"✅ Caso creado: {caso.caso_id}")

---

Consultar Casos

Puedes realizar búsquedas utilizando el módulo caso:

casos = client.caso().search()

print(casos)

También puedes trabajar con filtros y paginación disponibles en el SDK.

Consulta la documentación de Casos para conocer las operaciones disponibles.

---

Autenticación Automática

El SDK gestiona automáticamente el ciclo de vida del token de autenticación.

Al realizar una operación:

casos = client.caso().search()

el SDK:

1. Obtiene un token cuando es necesario.
2. Reutiliza el token mientras sea válido.
3. Renueva el token antes de su expiración.
4. Gestiona respuestas 401 según la lógica de autenticación.
5. Reintenta la operación cuando corresponde.

No es necesario realizar manualmente una llamada al endpoint /auth para cada operación.

Para conocer en detalle este comportamiento, consulta Autenticación y Ciclo del Token.

---

Uso Básico con Context Manager (with)

Ideal para scripts puntuales, jobs por lotes, CLI tools y pruebas unitarias/integración. Cierra automáticamente el pool HTTP e hilos internos al finalizar.

import os

from shori_sdk import ShoriClientBuilder, CaseFilter, FilterOperator

with (
    ShoriClientBuilder()
    .environment(os.environ.get("SHORI_ENVIRONMENT", "UAT"))
    .portal_id(os.environ["SHORI_PORTAL_ID"])
    .tenant_id(os.environ["SHORI_TENANT_ID"])
    .api_key(os.environ["SHORI_API_KEY"])
    .timeout_ms(30_000)
    .max_workers(8)
    .build()
) as client:
    result = client.caso().search(
        CaseFilter.builder()
        .caso_type_id(FilterOperator.EQ, "tipo-uuid")
        .working_sub_state_id(FilterOperator.IN, "estado-1", "estado-2")
        .page_size(20)
        .build()
    )

    print(f"Total encontrados: {result.count}")

    for caso in result.caso_response:
        print(f"ID: {caso.caso_id} - Número: {caso.caso_number}")

---

Uso Avanzado con Multi-Threading (Múltiples Hilos Concurrentes)

El cliente de Shori es completamente thread-safe. Múltiples hilos pueden compartir la misma instancia de ShoriClient para consultar la API de forma paralela sin duplicar solicitudes de autenticación ni saturar el pool.

import os

from concurrent.futures import ThreadPoolExecutor, as_completed

from shori_sdk import ShoriClientBuilder, CaseFilter, FilterOperator


def procesar_caso_id(client, caso_id: str):
    """Función ejecutada por cada hilo worker."""
    caso = client.caso().find_by_id(caso_id)

    # Operaciones asociadas al caso...
    return f"Procesado caso {caso.caso_number}"


# Se instancia 1 solo cliente compartido
client = (
    ShoriClientBuilder()
    .environment(os.environ["SHORI_ENVIRONMENT"])
    .portal_id(os.environ["SHORI_PORTAL_ID"])
    .tenant_id(os.environ["SHORI_TENANT_ID"])
    .api_key(os.environ["SHORI_API_KEY"])
    .max_workers(16)
    .build()
)

try:
    casos_ids = ["uuid-1", "uuid-2", "uuid-3", "uuid-4", "uuid-5"]

    # Ejecución paralela usando un ThreadPoolExecutor externo
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = [
            executor.submit(procesar_caso_id, client, cid)
            for cid in casos_ids
        ]

        for future in as_completed(futures):
            print(future.result())

finally:
    # Siempre cerrar el cliente al finalizar la aplicación multihilo
    client.close()

---

Integración con FastAPI

FastAPI ejecuta los endpoints declarados con def síncrono directamente en su propio threadpool interno. Puedes inyectar una instancia singleton gestionada en el evento lifespan.

import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Depends, HTTPException, Request

from shori_sdk import (
    ShoriClient,
    ShoriClientBuilder,
    ResourceNotFoundException,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Crear la instancia compartida en el arranque de FastAPI
    app.state.shori = (
        ShoriClientBuilder()
        .environment(os.environ["SHORI_ENVIRONMENT"])
        .portal_id(os.environ["SHORI_PORTAL_ID"])
        .tenant_id(os.environ["SHORI_TENANT_ID"])
        .api_key(os.environ["SHORI_API_KEY"])
        .build()
    )

    yield

    # Liberar conexiones y pools al apagar la aplicación
    app.state.shori.close()


app = FastAPI(title="Mi Servicio Shori", lifespan=lifespan)


def get_shori(request: Request) -> ShoriClient:
    return request.app.state.shori


# Endpoint síncrono
# FastAPI lo ejecuta automáticamente en un thread separado
@app.get("/casos/{caso_id}")
def obtener_caso(
    caso_id: str,
    client: ShoriClient = Depends(get_shori),
):
    try:
        return client.caso().find_by_id(caso_id)
    except ResourceNotFoundException:
        raise HTTPException(
            status_code=404,
            detail="Caso no encontrado",
        )


# Endpoint asíncrono usando asyncio.to_thread
# para no bloquear el Event Loop
@app.get("/casos/{caso_id}/async")
async def obtener_caso_async(
    caso_id: str,
    client: ShoriClient = Depends(get_shori),
):
    try:
        return await asyncio.to_thread(
            client.caso().find_by_id,
            caso_id,
        )
    except ResourceNotFoundException:
        raise HTTPException(
            status_code=404,
            detail="Caso no encontrado",
        )

---

Integración con Flask

Implementa un Singleton thread-safe mediante lru_cache y registra el cierre con atexit.

# shori_provider.py

import atexit
import os
from functools import lru_cache

from shori_sdk import ShoriClient, ShoriClientBuilder


@lru_cache(maxsize=1)
def get_shori_client() -> ShoriClient:
    client = (
        ShoriClientBuilder()
        .environment(os.environ["SHORI_ENVIRONMENT"])
        .portal_id(os.environ["SHORI_PORTAL_ID"])
        .tenant_id(os.environ["SHORI_TENANT_ID"])
        .api_key(os.environ["SHORI_API_KEY"])
        .build()
    )

    atexit.register(client.close)

    return client

# app.py

from flask import Flask, jsonify, abort

from shori_provider import get_shori_client
from shori_sdk import ResourceNotFoundException


app = Flask(__name__)


@app.route("/casos/<caso_id>", methods=["GET"])
def ver_caso(caso_id):
    client = get_shori_client()

    try:
        caso = client.caso().find_by_id(caso_id)
        return jsonify(caso.model_dump(by_alias=True))

    except ResourceNotFoundException:
        abort(404, description="Caso no encontrado")


if __name__ == "__main__":
    app.run(port=5000)

---

Integración con Django

Utiliza el mismo proveedor Singleton dentro de tus vistas de Django.

# views.py

from django.http import JsonResponse, Http404

from shori_provider import get_shori_client
from shori_sdk import ResourceNotFoundException


def detalle_caso(request, caso_id):
    client = get_shori_client()

    try:
        caso = client.caso().find_by_id(caso_id)

        return JsonResponse(
            caso.model_dump(by_alias=True)
        )

    except ResourceNotFoundException:
        raise Http404("El caso especificado no existe.")

---

Integración con Celery (Procesamiento Asíncrono / Background Workers)

Para trabajadores Celery, se recomienda instanciar el cliente a nivel de la tarea o vincular el cierre al evento de finalización del worker.

import os

from celery import Celery

from shori_sdk import ShoriClientBuilder


app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
)


@app.task(bind=True, max_retries=3)
def procesar_comentario_tarea(
    self,
    caso_id: str,
    mensaje: str,
):
    with (
        ShoriClientBuilder()
        .environment(os.environ["SHORI_ENVIRONMENT"])
        .portal_id(os.environ["SHORI_PORTAL_ID"])
        .tenant_id(os.environ["SHORI_TENANT_ID"])
        .api_key(os.environ["SHORI_API_KEY"])
        .build()
    ) as client:

        try:
            client.caso().comment().add(
                {
                    "casoId": caso_id,
                    "comment": mensaje,
                }
            )

            return f"Comentario agregado con éxito a {caso_id}"

        except Exception as exc:
            raise self.retry(
                exc=exc,
                countdown=10,
            )

---

Siguiente paso

Una vez completada la instalación, puedes continuar con:

• Autenticación y Ciclo del Token
• Módulo Caso
• Módulo Repositorio