Shori SDK para Python

Integra aplicaciones Python con la plataforma Syntpony Process Management (Shori).

Características

• ✅ httpx con pool de conexiones y soporte thread-safe
• ✅ Auto-renovación del token Bearer: una sola autenticación aunque haya N hilos concurrentes
• ✅ Builders fluidos + validación con Pydantic v2
• ✅ Filtros y requests inmutables: se pueden compartir entre hilos
• ✅ Descarga de adjuntos en paralelo con pool acotado mediante max_workers
• ✅ Tipado completo mediante py.typed
• ✅ Python 3.10+
• ✅ Compatible con scripts, automatizaciones y frameworks web como FastAPI, Flask y Django

Contenido

1. Instalación
2. Variables de entorno
3. Inicio rápido
4. Ciclo de vida del cliente
5. Uso en scripts y automatizaciones
6. Uso con frameworks web
7. Uso con base de datos
8. Hilos (multi-threading)
9. Ejemplos de la API
10. Manejo de errores
11. Equivalencias TypeScript → Python
12. Estructura del proyecto
13. Pruebas

---

Instalación

Este proyecto utiliza un entorno virtual para aislar las dependencias del SDK.

Crear el entorno virtual

Desde la raíz del proyecto shori-api-client-py:

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):

(.venv) PS C:\ruta\shori-api-client-py>

Verificar el intérprete activo

Comprueba que Python corresponde al entorno virtual:

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

La ruta debe apuntar a:

...\shori-api-client-py\.venv\Scripts\python.exe

También puedes verificar pip:

python -m pip --version

La ruta debe corresponder al mismo entorno virtual:

...\shori-api-client-py\.venv\Lib\site-packages

Nota: se recomienda utilizar python -m pip en lugar de pip directamente para garantizar que pip pertenece al mismo intérprete de Python activo.

Instalar el proyecto en modo editable

Una vez activado el entorno virtual:

python -m pip install -e .

La instalación editable permite modificar el código fuente y probar los cambios sin tener que reinstalar el paquete después de cada modificación.

Instalar dependencias de desarrollo

Para instalar también las herramientas utilizadas durante el desarrollo y testing:

python -m pip install -e ".[dev]"

Esto instala las dependencias definidas en el extra dev, incluyendo:

• pytest
• python-dotenv
• pyright

VS Code y Pylance

Si VS Code muestra errores como:

Import "httpx" could not be resolved

o marca clases del SDK como unknown, comprueba que esté utilizando el mismo entorno virtual.

En VS Code:

Ctrl + Shift + P → Python: Select Interpreter

Selecciona:

.venv\Scripts\python.exe

Si el problema persiste:

Ctrl + Shift + P → Developer: Reload Window

---

Variables de entorno

Los ejemplos y las pruebas de integración utilizan variables de entorno para evitar colocar credenciales directamente en el código.

Puedes utilizar un archivo .env a partir de .env.example:

cp .env.example .env

En Windows PowerShell:

Copy-Item .env.example .env

Ejemplo:

SHORI_PORTAL_ID=tu-portal-uuid
SHORI_TENANT_ID=tu-tenant-uuid
SHORI_API_KEY=tu-api-key
SHORI_ENVIRONMENT=UAT

Los valores disponibles para SHORI_ENVIRONMENT son:

DEV
UAT
PROD

Importante: el SDK no carga automáticamente estas variables de entorno. Los ejemplos y las pruebas las leen explícitamente mediante os.environ o python-dotenv y posteriormente las pasan al ShoriClientBuilder.

Nunca debes versionar credenciales reales ni tu archivo .env.

---

Inicio rápido

El siguiente ejemplo crea un cliente, realiza una búsqueda y libera automáticamente los recursos internos mediante with:

import os

from shori_sdk import (
    CaseFilter,
    FilterOperator,
    ShoriClientBuilder,
)

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"])
    .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} - "
            f"Número: {caso.caso_number}"
        )

---

Ciclo de vida del cliente

La idea es similar a un bean singleton de Spring: una instancia de ShoriClient por aplicación o proceso, reutilizada durante su ciclo de vida.

Spring Boot	Python
@Bean singleton	Una instancia mediante variable de módulo, lru_cache o lifespan del framework
@PreDestroy / destroyMethod = "close"	with, atexit o evento de shutdown
@Autowired	Depends(...) en FastAPI o proveedor compartido en Flask/Django

with: scripts, jobs y tests

with llama automáticamente a close() al salir del bloque, incluso si ocurre una excepción.

Úsalo cuando el proceso nace, trabaja y termina:

from shori_sdk import ShoriClientBuilder

with (
    ShoriClientBuilder()
    .environment("UAT")
    .portal_id("tu-portal-id")
    .tenant_id("tu-tenant-id")
    .api_key("tu-api-key")
    .build()
) as client:

    client.caso().search()

⚠️ No uses with por request en un servidor web. Esto crearía un cliente, un pool HTTP y un ciclo de autenticación nuevos en cada llamada.

Singleton para aplicaciones de larga vida

Una opción sencilla es utilizar lru_cache para crear el cliente una sola vez:

# 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

close() es idempotente y libera los recursos internos del cliente.

---

Uso en scripts y automatizaciones

Para automatizaciones no se necesita un framework web. Puedes ejecutar un script desde un scheduler, una cola, un proceso batch o cualquier otro mecanismo de automatización.

Script puntual

import logging
import os

from shori_sdk import (
    CaseFilter,
    FilterOperator,
    NextStateRequest,
    ShoriClientBuilder,
)

logging.basicConfig(level=logging.INFO)


def main() -> None:

    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:

        casos = client.caso().search_all(
            CaseFilter.builder()
            .caso_type_id(
                FilterOperator.EQ,
                "tipo-uuid",
            )
            .working_sub_state_id(
                FilterOperator.EQ,
                "estado-pendiente",
            )
            .build()
        )

        for caso in casos:
            client.caso().state().next(
                NextStateRequest(
                    caso_id=caso.caso_id,
                    working_sub_state_primary_level_id="destino",
                )
            )


if __name__ == "__main__":
    main()

Procesar en paralelo

Una misma instancia de ShoriClient puede compartirse entre múltiples hilos:

import logging

from concurrent.futures import ThreadPoolExecutor, as_completed


def procesar(caso):
    client.caso().comment().add(
        {
            "casoId": caso.caso_id,
            "comment": "Procesado por bot",
        }
    )

    return caso.caso_id


with ThreadPoolExecutor(max_workers=8) as pool:

    futuros = {
        pool.submit(procesar, caso): caso
        for caso in casos
    }

    for future in as_completed(futuros):
        caso = futuros[future]

        try:
            future.result()
        except Exception:
            logging.exception(
                "Falló el caso %s",
                caso.caso_id,
            )

Cómo dispararlo

Necesidad	Herramienta
Cada X minutos u horas	cron, Programador de tareas de Windows, systemd timer
Programar dentro de Python	APScheduler
Servicio que corre continuamente	while True + time.sleep
Reintentos, colas y múltiples workers	Celery o RQ
Flujos con dependencias, monitoreo y UI	Prefect o Airflow
Reaccionar a eventos	FastAPI mediante webhooks
Automatizar navegador o escritorio	Playwright, Selenium, Robot Framework

Job periódico con APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler

from shori_sdk import ShoriClientBuilder


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


def revisar():
    client.caso().search()


scheduler = BlockingScheduler()

scheduler.add_job(
    revisar,
    "interval",
    minutes=5,
    max_instances=1,
)

try:
    scheduler.start()
finally:
    client.close()

max_instances=1 evita que dos ejecuciones de la misma tarea se ejecuten simultáneamente si una tarda más de lo esperado.

---

Uso con frameworks web

El SDK es síncrono (bloqueante) y thread-safe. Está pensado para servidores que ejecutan las operaciones en pools de hilos.

FastAPI

lifespan puede utilizarse para gestionar el ciclo de vida del cliente.

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, Request

from shori_sdk import ShoriClient, ShoriClientBuilder


@asynccontextmanager
async def lifespan(app: FastAPI):

    app.state.shori = (
        ShoriClientBuilder()
        .environment("UAT")
        .portal_id("tu-portal-id")
        .tenant_id("tu-tenant-id")
        .api_key("tu-api-key")
        .build()
    )

    yield

    app.state.shori.close()


app = FastAPI(lifespan=lifespan)


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


@app.get("/casos/{caso_id}")
def get_caso(
    caso_id: str,
    client: ShoriClient = Depends(shori),
):
    return client.caso().find_by_id(caso_id)

Para endpoints definidos con def, FastAPI puede ejecutarlos en su pool de hilos.

Si necesitas utilizar async def, evita bloquear el event loop:

import asyncio

from fastapi import FastAPI

app = FastAPI()


@app.get("/casos/{caso_id}/async")
async def get_caso_async(
    caso_id: str,
    client: ShoriClient = Depends(shori),
):
    return await asyncio.to_thread(
        client.caso().find_by_id,
        caso_id,
    )

Flask

from flask import Flask, jsonify

from shori_provider import get_shori_client


app = Flask(__name__)


@app.get("/casos/<caso_id>")
def get_caso(caso_id):

    caso = get_shori_client().caso().find_by_id(caso_id)

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

Django

# views.py

from django.http import JsonResponse

from shori_provider import get_shori_client


def caso_detail(request, caso_id):

    caso = get_shori_client().caso().find_by_id(caso_id)

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

Varios workers

Cuando una aplicación utiliza múltiples procesos, como Gunicorn o Uvicorn con varios workers, cada proceso tendrá su propia instancia de ShoriClient, con su propio pool y ciclo de autenticación.

Los procesos no comparten memoria.

Dentro de cada proceso, la instancia puede compartirse entre sus hilos.

---

Uso con base de datos

Para automatizaciones que necesitan trabajar con una base de datos, una alternativa habitual es SQLAlchemy 2.

Necesidad	Herramienta
Pool / ORM	SQLAlchemy 2
PostgreSQL	psycopg
SQL Server	pyodbc
Oracle	oracledb
MySQL / MariaDB	PyMySQL o mysqlclient
SQLite	sqlite3
Migraciones	Alembic

Engine

El Engine de SQLAlchemy funciona como el punto central de administración del pool de conexiones:

import os

from sqlalchemy import create_engine, text


engine = create_engine(
    os.environ["DATABASE_URL"],
    pool_size=5,
    max_overflow=5,
    pool_pre_ping=True,
    pool_recycle=1800,
)

with engine.begin() as conn:

    filas = conn.execute(
        text(
            "SELECT id, caso_uuid "
            "FROM pendientes "
            "WHERE estado = :estado"
        ),
        {"estado": "NUEVO"},
    ).all()

engine.dispose()

Shori + base de datos

import os

from concurrent.futures import ThreadPoolExecutor

from sqlalchemy import create_engine, text

from shori_sdk import ShoriClientBuilder


def main() -> None:

    engine = create_engine(
        os.environ["DATABASE_URL"],
        pool_size=8,
        pool_pre_ping=True,
    )

    try:

        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 shori:

            with engine.begin() as conn:

                pendientes = conn.execute(
                    text(
                        "SELECT id, caso_uuid "
                        "FROM pendientes"
                    )
                ).all()

            def procesar(fila):

                shori.caso().comment().add(
                    {
                        "casoId": fila.caso_uuid,
                        "comment": "Procesado",
                    }
                )

                with engine.begin() as conn:

                    conn.execute(
                        text(
                            "UPDATE pendientes "
                            "SET estado = 'OK' "
                            "WHERE id = :id"
                        ),
                        {"id": fila.id},
                    )

            with ThreadPoolExecutor(max_workers=8) as pool:
                list(pool.map(procesar, pendientes))

    finally:
        engine.dispose()


if __name__ == "__main__":
    main()

Reglas con hilos

• El Engine puede compartirse entre hilos; las conexiones y sesiones no.
• Cada hilo debe obtener su propia conexión mediante engine.begin() o crear su propia Session.
• Mantén max_workers acorde al tamaño del pool de conexiones para evitar esperas innecesarias.
• Cierra siempre los recursos con with o try/finally.
• Usa mecanismos de idempotencia para evitar duplicar operaciones ante reintentos.
• Mantén las credenciales fuera del código, utilizando variables de entorno o un gestor de secretos.

---

Hilos (multi-threading)

Crea una sola instancia de ShoriClient y compártela entre los hilos.

Pieza	Garantía
httpx.Client	Pool de conexiones compartido y reutilización de conexiones
TokenProvider	Sincroniza la obtención y renovación del token
Autenticación concurrente	Múltiples hilos sin token pueden compartir una misma autenticación
Retry en 401	La renovación se coordina para evitar renovaciones innecesarias
ShoriConfig, filtros y requests	Inmutables y compartibles
Responses	No deben compartirse entre hilos mientras estén siendo modificadas
Builders	Mutables y no thread-safe; utiliza un builder por llamada o hilo

Ejemplo:

from concurrent.futures import ThreadPoolExecutor

from shori_sdk import (
    CaseFilter,
)


shared_filter = (
    CaseFilter.builder()
    .working_sub_state_id("estado")
    .build()
)


with ThreadPoolExecutor(max_workers=8) as pool:

    results = list(
        pool.map(
            lambda _: client.caso().search_all(shared_filter),
            range(20),
        )
    )

Las operaciones que realizan descargas en paralelo utilizan el pool interno del cliente:

• get_attachments
• search_with_attachments
• search_all_with_attachments

El número máximo de workers se controla con max_workers, cuyo valor por defecto es 8.

---

Ejemplos de la API

from shori_sdk import (
    AddCommentRequest,
    CaseFilter,
    CasoCreateRequest,
    CommentFilter,
    CreateCasoMassiveRequest,
    FilterOperator,
    GetDocumentByIdRequest,
    SortDirection,
    UploadDocumentRequest,
)


# Crear un caso
res = client.caso().create(
    CasoCreateRequest.builder()
    .caso_type_id("tipo-uuid")
    .form_id("form-uuid")
    .priority("prioridad-uuid")
    .submitted_data(
        {
            "nombreCliente": "Ana García",
        }
    )
    .build()
)

print(res.caso_id, res.caso_number)


# Creación masiva
client.caso().create_massive(
    CreateCasoMassiveRequest.builder()
    .caso_type_id("producto-uuid")
    .add_caso(
        CasoCreateRequest.builder()
        .caso_type_id("t")
        .form_id("f")
        .priority("p")
        .build()
    )
    .update_duplicates(True)
    .build()
)


# Filtros
filtro = (
    CaseFilter.builder()
    .caso_type_id(
        FilterOperator.EQ,
        "tipo-uuid",
    )
    .working_sub_state_id(
        "estado-1",
        "estado-2",
    )
    .data_like(
        "nombre",
        "García",
    )
    .created_at(
        "2024-01-01",
        "2024-12-31",
        "America/Lima",
    )
    .page_size(50)
    .order_by("createdAt")
    .sort_direction(SortDirection.DESC)
    .build()
)

una_pagina = client.caso().search(filtro)

todos = client.caso().search_all(filtro)

con_adjuntos = client.caso().search_all_with_attachments(filtro)


# Comentarios
client.caso().comment().add(
    AddCommentRequest.builder()
    .caso_id("uuid")
    .comment("texto")
    .build()
)

comentarios = client.caso().comment().search_all(
    CommentFilter.builder()
    .caso_id("uuid")
    .build()
)


# Estados
client.caso().state().next(
    {
        "casoId": "uuid",
        "workingSubStatePrimaryLevelId": "estado-destino",
    }
)


# Documentos
doc_id = client.repo().document().upload(
    UploadDocumentRequest(
        file="contrato.pdf",
        caso_id="uuid",
        caso_type_id="tipo-uuid",
    )
)

archivo = client.repo().download().by_document(
    GetDocumentByIdRequest(
        document_id=doc_id,
        file_name="contrato.pdf",
    )
)

archivo.save("/descargas")

---

Manejo de errores

from pydantic import ValidationError

from shori_sdk import (
    ApiException,
    AuthenticationException,
    ResourceNotFoundException,
    ValidationException,
)


try:

    client.caso().find_by_id("uuid")

except ResourceNotFoundException:
    # 404
    ...

except AuthenticationException:
    # credenciales inválidas o 401 después del flujo de autenticación
    ...

except ApiException as e:
    # otros errores HTTP o de red
    print(e.status_code)

Además:

• pydantic.ValidationError: datos inválidos al construir modelos o requests.
• ValidationException: parámetros inválidos detectados por el SDK, por ejemplo timeout_ms <= 0.

---

Equivalencias TypeScript → Python

TypeScript	Python
camelCase en métodos/campos	snake_case, por ejemplo casoTypeId() → caso_type_id()
JSON del API en camelCase	Se mantiene mediante aliases de Pydantic
client.caso().casoType()	client.caso().caso_type()
Zod schema + .parse()	Modelo Pydantic + validación en build()
z.ZodError	pydantic.ValidationError
interface XRequest	Clase Pydantic o dict
File / Blob	ShoriFile
Promise<T> / async	Operaciones síncronas + hilos
Promise.all	Paralelización mediante helpers internos
URLSearchParams	urlencode
FilterOperator	FilterOperator(Enum)
JSON.stringify omite undefined	El serializador maneja None según la configuración del modelo

Entre las mejoras respecto a la implementación TypeScript se incluyen:

• get_attachments evita duplicar archivos o reenviar páginas.
• search_all utiliza paginación iterativa.
• AuthService utiliza AuthenticationException.
• Las operaciones update() y upload() no mutan el objeto recibido.

---

Estructura del proyecto

shori-api-client-py/

├── pyproject.toml
├── README.md
├── .env.example
│
├── src/                         ← src layout
│   └── shori_sdk/               ← paquete real
│       ├── __init__.py          ← API pública
│       ├── auth/
│       ├── client/
│       ├── config/
│       ├── exception/
│       ├── http/
│       ├── utils/
│       │
│       └── modules/
│           ├── caso/
│           └── repo/
│
└── tests/
    ├── unit/                    ← pruebas sin red
    └── integration/             ← pruebas contra UAT

src layout

El proyecto utiliza el patrón src layout. El código del paquete se encuentra en:

src/shori_sdk/

Esto ayuda a que las pruebas utilicen el paquete instalado y permite detectar problemas de empaquetado antes de publicar una nueva versión.

shori-sdk vs shori_sdk

• shori-sdk: nombre de distribución utilizado para instalar el paquete con pip.
• shori_sdk: nombre del paquete utilizado en los imports de Python.

Ejemplo:

python -m pip install shori-sdk

import shori_sdk

Archivos __init__.py

Cada paquete de Python contiene su correspondiente __init__.py.

El __init__.py principal de shori_sdk define la API pública del SDK y permite imports como:

from shori_sdk import ShoriClientBuilder

Archivos *.egg-info

Los directorios *.egg-info/ pueden generarse al instalar el proyecto en modo editable.

Contienen metadatos de instalación y no deben editarse manualmente ni versionarse en Git.

---

Pruebas

Las pruebas se encuentran separadas entre pruebas unitarias y pruebas de integración.

Dependencias de desarrollo

Si todavía no instalaste las dependencias de desarrollo:

python -m pip install -e ".[dev]"

Pruebas unitarias

Las pruebas unitarias no requieren conexión con Shori:

pytest -m "not integration" -v

Pyright

El proyecto utiliza Pyright en modo strict para validar el código fuente:

pyright

Pruebas de integración

Las pruebas de integración utilizan el entorno UAT y requieren credenciales válidas.

Primero configura tu .env:

SHORI_PORTAL_ID=tu-portal-uuid
SHORI_TENANT_ID=tu-tenant-uuid
SHORI_API_KEY=tu-api-key
SHORI_ENVIRONMENT=UAT

Después ejecuta:

pytest -m integration -v

Las credenciales utilizadas para las pruebas de integración deben mantenerse fuera del repositorio. El archivo .env no debe versionarse.

---

Jerson Omar Ramírez Ortiz