Shori SDK para Python

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

Características

• ✅ httpx con pool de conexiones, thread-safe
• ✅ Auto-renovación del token Bearer: una sola autenticación aunque haya N hilos
• ✅ Builders fluidos + validación con pydantic v2 (equivalente a Zod)
• ✅ Filtros y requests inmutables: se comparten entre hilos sin locks
• ✅ Descarga de adjuntos en paralelo con pool acotado (max_workers)
• ✅ Tipado completo (py.typed) · Python ≥ 3.10

Contenido

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

---

Instalación

pip install -e .            # desarrollo
pip install -e ".[dev]"     # + pytest y python-dotenv

Entorno virtual y VS Code (Pylance)

Si el editor marca Import "httpx" could not be resolved, VS Code está usando un Python distinto al que tiene las dependencias instaladas. Crea un entorno virtual en el proyecto:

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -e ".[dev]"

Luego en VS Code: Ctrl+Shift+P → Python: Select Interpreter → elige .venv (y Developer: Reload Window si el error persiste). El código de src/ pasa pyright en modo strict (configurado en pyproject.toml); los tests se analizan en modo básico.

Variables de entorno recomendadas (ver .env.example; nunca commitees tu .env):

SHORI_PORTAL_ID=tu-portal-uuid
SHORI_TENANT_ID=tu-tenant-uuid
SHORI_API_KEY=tu-api-key
SHORI_ENVIRONMENT=UAT        # DEV | UAT | PROD

Instalación

Instala el SDK directamente con pip:

pip install shori-api-client-py

Para instalar una versión específica:

pip install shori-api-client-py==1.0.0

Puedes verificar la instalación con:

pip show shori-api-client-py

---

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:

.venv\Scripts\activate

Una vez activado el entorno, instala el SDK:

pip install shori-api-client-py

---

Inicio rápido

import os
from shori_sdk import ShoriClientBuilder, CaseFilter, FilterOperator

with (
    ShoriClientBuilder()
    .environment(os.environ["SHORI_ENVIRONMENT"])   # ShoriEnvironment.PROD o "PROD"
    .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)                             # opcional (30 s por defecto)
    .max_workers(8)                                 # opcional: hilos internos (8 por defecto)
    .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(result.count, [c.caso for c in result.caso_response])

---

Ciclo de vida del cliente

Es la misma idea que un bean singleton de Spring: un solo cliente por aplicación, creado al arrancar y cerrado al apagar.

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

with: scripts, jobs y tests

with llama a close() al salir del bloque, incluso si hay una excepción (equivale a try/finally). Úsalo cuando el proceso nace, trabaja y termina.

⚠️ No uses with por request en un servidor: crearía un pool y un token nuevos en cada llamada.

Singleton para apps de larga vida

# shori_provider.py — equivalente a tu @Configuration
import atexit
import os
from functools import lru_cache

from shori_sdk import ShoriClient, ShoriClientBuilder


@lru_cache(maxsize=1)            # se crea la primera vez y se reutiliza
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)   # el "@PreDestroy" para frameworks sin hook propio
    return client

close() es idempotente y libera el pool de hilos y las conexiones HTTP.

---

Uso en scripts y automatizaciones

Para automatizar no se necesita un framework web: se escribe un script y algo externo lo dispara.

Script puntual

import logging
import os
from shori_sdk import ShoriClientBuilder, CaseFilter, FilterOperator, NextStateRequest

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 (mismo cliente en todos los hilos)

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, c): c for c in casos}
    for f in as_completed(futuros):
        try:
            f.result()
        except Exception:
            logging.exception("Falló el caso %s", futuros[f].caso_id)   # un fallo no tumba al resto

Cómo dispararlo

Necesidad	Herramienta
Cada X minutos u horas	cron (Linux), Programador de tareas (Windows), systemd timer
Programar dentro de Python	APScheduler
Servicio que corre siempre (polling)	while True + time.sleep, gestionado por systemd o Docker
Reintentos, colas, muchos workers	Celery o RQ
Flujos con dependencias, monitoreo y UI	Prefect o Airflow
Reaccionar a un evento (webhook)	FastAPI (ver abajo)
Automatizar navegador / escritorio	Playwright, Selenium, Robot Framework (el SDK se usa dentro igual)

Job periódico con APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler

client = ShoriClientBuilder()....build()          # una vez, al arrancar (como el bean)
scheduler = BlockingScheduler()
scheduler.add_job(lambda: revisar(client), "interval", minutes=5, max_instances=1)

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

max_instances=1 evita que dos ejecuciones se pisen si una tarda más de 5 minutos.

---

Uso con frameworks web

El SDK es síncrono (bloqueante) y thread-safe, pensado para servidores que atienden requests en un pool de hilos.

FastAPI (lifespan + Depends)

Lo más parecido a Spring: lifespan = ciclo de vida del bean, Depends = inyección.

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()....build()   # arranque = crear bean
    yield
    app.state.shori.close()                              # apagado = @PreDestroy


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)):   # def, NO async def
    return client.caso().find_by_id(caso_id)

Declara los endpoints con def: FastAPI los ejecuta en un pool de hilos. Si necesitas async def, no bloquees el event loop:

import asyncio

@app.get("/casos/{caso_id}")
async def get_caso(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   # se cierra con atexit

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 (gunicorn / uvicorn)

Cada proceso worker tiene su propia instancia del cliente (token y pool propios), porque los procesos no comparten memoria. Es lo normal. Lo que se comparte es la instancia entre los hilos de un mismo proceso, y eso es lo que el SDK garantiza.

---

Uso con base de datos

Python no trae un "Spring Data" integrado. Para automatizaciones lo más práctico es SQLAlchemy 2 (pool + reconexión + todos los motores cambiando solo la URL).

Necesidad	Herramienta
Pool / ORM (≈ JPA + HikariCP)	SQLAlchemy 2
PostgreSQL	psycopg (v3)
SQL Server	pyodbc (requiere el driver ODBC instalado)
Oracle	oracledb
MySQL / MariaDB	PyMySQL o mysqlclient
SQLite (local, pruebas)	sqlite3 (incluido)
Migraciones (Flyway/Liquibase)	Alembic

Engine = tu DataSource

import os
from sqlalchemy import create_engine, text

engine = create_engine(
    os.environ["DATABASE_URL"],   # postgresql+psycopg://user:pass@host/db
    pool_size=5,                  # conexiones fijas
    max_overflow=5,               # extra en picos
    pool_pre_ping=True,           # detecta conexiones muertas
    pool_recycle=1800,            # recicla cada 30 min
)

with engine.begin() as conn:      # = @Transactional: commit al salir, rollback si hay excepción
    filas = conn.execute(text("SELECT id, caso_uuid FROM pendientes WHERE estado = :e"), {"e": "NUEVO"}).all()

# al terminar el programa:
engine.dispose()                  # = close() del pool

Shori + base de datos en un mismo script

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:   # cada hilo toma SU conexión del pool
                    conn.execute(text("UPDATE pendientes SET estado='OK' WHERE id=:i"), {"i": 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 se comparte entre hilos; las conexiones y sesiones no. Cada hilo abre la suya con engine.begin() o Session(engine).
• Mantén max_workers ≤ pool_size + max_overflow, o los hilos sobrantes esperan conexión y pueden dar timeout.
• Cierra siempre con with / try-finally: una conexión sin devolver al pool queda ocupada.
• Idempotencia: marca en la base qué casos ya procesaste, para que una doble ejecución no duplique comentarios ni cambios de estado.
• Credenciales en variables de entorno o gestor de secretos, nunca en el código. Para reintentos usa tenacity.

---

Hilos (multi-threading)

Crea un solo ShoriClient y compártelo entre todos los hilos.

Pieza	Garantía
httpx.Client	thread-safe; un pool de conexiones keep-alive compartido
TokenProvider	lock + single-flight: N hilos sin token → 1 autenticación
Retry en 401	cada request informa qué token usó; si otro hilo ya renovó, se reutiliza el nuevo (N hilos con 401 → 1 renovación, no N)
ShoriConfig, filtros, requests	inmutables (frozen) → compartibles
Responses	mutables (como en TS: search() remapea caso_id/caso_number): no los compartas entre hilos mientras los modificas
Builders	mutables, no thread-safe → un builder por llamada/hilo

from concurrent.futures import ThreadPoolExecutor

shared_filter = CaseFilter.builder().working_sub_state_id("estado").build()   # inmutable
with ThreadPoolExecutor(8) as pool:
    results = list(pool.map(lambda _: client.caso().search_all(shared_filter), range(20)))

Operaciones que paralelizan internamente (equivalentes a Promise.all del TS): get_attachments, search_with_attachments y search_all_with_attachments. Usan el pool acotado del cliente (max_workers, 8 por defecto) y devuelven los resultados en orden. No hay riesgo de deadlock con pools pequeños (probado con max_workers=1).

---

Ejemplos de la API

from shori_sdk import (
    AddCommentRequest, AttachmentFilesFilter, CasoCreateRequest, CaseFilter, 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: operador explícito, IN implícito, campos del formulario y rango de fechas
filtro = (
    CaseFilter.builder()
    .caso_type_id(FilterOperator.EQ, "tipo-uuid")
    .working_sub_state_id("estado-1", "estado-2")            # → in: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)                     # auto-paginación
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 (también acepta dict en camelCase o snake_case)
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")          # también: archivo.content, archivo.size, archivo.type

---

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 / 401 tras renovar
except ApiException as e: print(e.status_code)   # 4xx/5xx; 408 = timeout; 0 = error de red

• pydantic.ValidationError: un builder/filtro con datos inválidos (equivale al ZodError).
• ValidationException: parámetros inválidos del SDK (ej. timeout_ms <= 0, AddCommentRequest sin texto).

---

Equivalencias TypeScript → Python

TypeScript	Python
camelCase en métodos/campos	snake_case (casoTypeId() → caso_type_id()); el JSON al API sigue en camelCase vía alias
client.caso().casoType()	client.caso().caso_type()
Zod schema + .parse() en build()	modelo pydantic + model_validate() en build()
z.ZodError	pydantic.ValidationError (mismos mensajes y límites)
interface XRequest (object literal)	clase pydantic o dict (camelCase o snake_case)
File / Blob	ShoriFile (acepta bytes, ruta o archivo binario abierto)
Promise<T> / async	síncrono + hilos
Promise.all	parallel_map (orden preservado, fail-fast)
URLSearchParams	urlencode
FilterOperator (clase singleton)	FilterOperator(Enum) con .apply() y has_operator()
JSON.stringify omite undefined	el serializador omite None

Mejoras respecto al TS: get_attachments ya no duplica archivos ni reenvía la misma página; search_all es iterativo (sin recursión); AuthService lanza AuthenticationException; update()/upload() ya no mutan el objeto recibido.

---

Estructura del proyecto

shori-api-client-py/
├── pyproject.toml
├── README.md
├── .env.example
├── src/                      ← contenedor ("src layout"): no es un paquete
│   └── shori_sdk/            ← el paquete real: `from shori_sdk import ...`
│       ├── __init__.py       ← API pública (equivale al index.ts)
│       ├── config/  auth/  http/  client/  exception/  utils/
│       └── modules/
│           ├── caso/         ← módulos, filter/, dto/request, dto/response
│           └── repo/         ← módulos y dto/
└── tests/
    ├── unit/                 ← sin red (servidor simulado)
    └── integration/          ← contra UAT (requieren credenciales)

• src/ layout: el código solo se puede importar si está instalado, así los tests prueban lo mismo que recibirá el usuario y se detectan errores de empaquetado antes de producción.
• shori-sdk vs shori_sdk: shori-sdk es el nombre de distribución utilizado para instalar el SDK mediante pip; shori_sdk es el nombre de importación utilizado dentro de Python.
• Varios __init__.py: cada carpeta con código necesita el suyo. Los vacíos marcan la carpeta como paquete (sin ellos, pip install . o el wheel omiten esas carpetas y falla con ModuleNotFoundError). Los que tienen contenido (shori_sdk/, dto/request/, dto/response/) re-exportan la API pública.
• *.egg-info/: lo genera pip install; son metadatos de instalación, no se edita ni se versiona (ya está en .gitignore).

---

Pruebas

pytest                     # unitarias (sin red, servidor simulado) — 42 tests
cp .env.example .env       # completar credenciales UAT
pytest -m integration      # equivalentes a los tests de integración del SDK TS

---

Jerson Omar Ramírez Ortiz