SAP: una sesión por conexión, reutilizada y renovada solo ante un 401

Antes se abría un login por orden (un cliente por orden, con la sesión
guardada solo en la instancia) y nunca se cerraba. Ahora el B1SESSION se
guarda por conexión (URL + base + usuario) en un caché del proceso, se
renueva solo si SAP responde 401 y se cierra con Logout al terminar el
proceso (SAPServiceLayerClient.cerrar_sesiones).

Probado contra SAP: 6 clientes de ambas tiendas hacen 1 login; una sesión
inválida provoca 1 login tras el 401; la integración de 3 órdenes con SAP
bloqueado para escrituras hace 1 login.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

core/sap.py

@@ -1,6 +1,8 @@
+import atexit
 import json
 import logging
-from typing import Any, Dict, Optional
+import threading
+from typing import Any, Dict, Optional, Tuple
 
 import requests
 
@@ -11,12 +13,19 @@ logger = logging.getLogger(__name__)
 
 class SAPServiceLayerClient:
     """
-    Cliente de SAP B1 Service Layer para una conexión (URL + CompanyDB)
-    específica. Cada tienda Shopify puede apuntar a una base SAP distinta,
-    por eso la sesión (B1SESSION) vive por instancia y no en un estado
-    global del módulo.
+    Cliente de SAP B1 Service Layer para una conexión (URL + CompanyDB +
+    usuario) específica. Cada tienda Shopify puede apuntar a una base SAP
+    distinta, por eso la sesión (B1SESSION) se guarda **por conexión** y se
+    reutiliza entre instancias del mismo proceso: crear un cliente por orden
+    no abre una sesión nueva por orden. Si dos tiendas comparten URL, base y
+    usuario, comparten sesión. La sesión se renueva sola ante un 401 y se
+    cierra con cerrar_sesiones() (también al terminar el proceso).
     """
 
+    # clave (url, company_db, username) -> {"headers", "url", "timeout", "verify_ssl"}
+    _sesiones: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
+    _lock = threading.Lock()
+
     def __init__(
         self,
         url: str,
@@ -34,6 +43,10 @@ class SAPServiceLayerClient:
         self.verify_ssl = verify_ssl
         self._headers: Optional[Dict[str, str]] = None
 
+    @property
+    def _clave(self) -> Tuple[str, str, str]:
+        return (self.url, self.company_db, self.username)
+
     @classmethod
     def from_store_config(cls, sap_connection: Dict[str, Any]) -> "SAPServiceLayerClient":
         """
@@ -81,10 +94,14 @@ class SAPServiceLayerClient:
     # ========================================================================
     def ensure_session(self, force_refresh: bool = False) -> Dict[str, str]:
         """
-        Verifica si existe una sesión (B1SESSION) para esta conexión.
-        Si no existe o force_refresh es True, hace Login.
+        Devuelve la sesión (B1SESSION) guardada para esta conexión. Si no
+        existe o force_refresh es True, hace Login y la guarda para las
+        próximas llamadas e instancias.
         """
-        if self._headers and not force_refresh:
+        with self._lock:
+            guardada = self._sesiones.get(self._clave)
+            if guardada and not force_refresh:
+                self._headers = guardada["headers"]
                 return self._headers
 
             logger.info(f"[{self.company_db}] Iniciando sesión en SAP Service Layer...")
@@ -109,6 +126,7 @@ class SAPServiceLayerClient:
                         f"Status: {response.status_code}"
                     )
                     self._headers = None
+                    self._sesiones.pop(self._clave, None)
                     raise SAPServiceLayerError(response.status_code, "Conexión fallida con el servidor SAP")
 
                 sap_session = response.json()
@@ -117,11 +135,13 @@ class SAPServiceLayerClient:
 
                 headers = {"Cookie": f"B1SESSION={b1_session_id}; ROUTEID={route_id}"}
                 self._headers = headers
+                self._sesiones[self._clave] = {"headers": headers, "url": self.url, "timeout": self.timeout, "verify_ssl": self.verify_ssl}
                 logger.info(f"[{self.company_db}] Sesión establecida. B1SESSION: {b1_session_id[:10]}...")
                 return headers
             except requests.RequestException as exc:
                 logger.error(f"[{self.company_db}] Error de conexión al autenticar contra SAP: {exc}")
                 self._headers = None
+                self._sesiones.pop(self._clave, None)
                 raise SAPServiceLayerError(500, f"Error al conectar con SAP Service Layer: {exc}")
 
     # ========================================================================
@@ -209,3 +229,32 @@ class SAPServiceLayerClient:
             raise SAPServiceLayerError(response.status_code, error_msg, code=error_code)
 
         return response
+
+    # ========================================================================
+    @classmethod
+    def cerrar_sesiones(cls) -> int:
+        """
+        Cierra (Logout) todas las sesiones guardadas y vacía el caché. Se llama
+        sola al terminar el proceso; devuelve cuántas sesiones se cerraron.
+        """
+        with cls._lock:
+            sesiones = list(cls._sesiones.items())
+            cls._sesiones.clear()
+
+        cerradas = 0
+        for (url, company_db, _), datos in sesiones:
+            try:
+                requests.post(
+                    f"{url}/Logout",
+                    headers={**datos["headers"], "Content-Type": "application/json"},
+                    timeout=datos["timeout"],
+                    verify=datos["verify_ssl"],
+                )
+                cerradas += 1
+                logger.info(f"[{company_db}] Sesión de SAP cerrada")
+            except requests.RequestException as exc:
+                logger.warning(f"[{company_db}] No se pudo cerrar la sesión de SAP (expirará sola): {exc}")
+        return cerradas
+
+
+atexit.register(SAPServiceLayerClient.cerrar_sesiones)

docs/flujo_pipeline.md

@@ -277,5 +277,8 @@ Este documento describe el flujo base. Las reglas siguientes lo complementan (de
 - **Idempotencia contra SAP:** antes de crear Invoice, ITR, StockTransfer o Delivery se busca un documento ya
   existente para la orden (`U_ShopifyOrderId` + `U_ShopifyStore`, y mismas líneas); si existe se adopta y no se
   duplica. Una Invoice ya pagada en SAP no genera otro pago.
+- **Sesión de SAP:** el `B1SESSION` se guarda por conexión (URL + base + usuario) y se reutiliza entre órdenes y
+  ciclos; solo se vuelve a hacer `Login` si SAP responde 401, y se hace `Logout` al terminar el proceso
+  (`SAPServiceLayerClient.cerrar_sesiones()`). Antes se abría una sesión por orden y nunca se cerraba.
 - **Red:** `ShopifyClient` reintenta hasta 4 veces ante fallos de red (SSL, conexión, timeout), con espera creciente.
 - **Cursor:** `max(id máximo en BD, initial_since_id)`.

docs/pendientes.md

@@ -77,6 +77,8 @@ Estado al 2026-09-30. Ver `docs/flujo_pipeline.md`, `docs/mapeo_campos.md` y
 - **Idempotencia contra SAP**: antes de crear Invoice, ITR, StockTransfer o Delivery se busca un documento
   existente de la orden (`U_ShopifyOrderId`, `U_ShopifyStore`, mismas líneas) y se adopta; una Invoice ya
   pagada no genera otro pago.
+- **Sesiones de SAP**: una por conexión, reutilizada y renovada solo ante un 401; `Logout` al cerrar el proceso
+  (antes: un login por orden, sin cierre).
 - **Reintentos de red** en `ShopifyClient` (4 intentos con espera creciente).
 - Invoice igual a lo cobrado (líneas canceladas o devueltas, reembolsos), fulfillments limitados a lo facturado,
   órdenes anuladas (`cancelled`), pedidos ignorados (`ignored`), despachadas y luego reembolsadas se integran.