""" Tablas de Ramaflex: las tres operaciones (listar, guardar, borrar) sobre las tablas del catálogo. Implementa ramaflex/pseudocodigo.txt. Especificación: https://docurm.random.cl/integracion-erp.html#tablas No depende de ningún framework HTTP. Uso: tablas = TablasRamaflex('sqlserver', lambda: pyodbc.connect(CADENA)) estado, cuerpo_json = tablas.atender('GET', 'XMOVILDOC', [('EMPRESA', '01')], '') `conectar` es una función sin argumentos que devuelve una conexión DB-API 2.0 nueva (pyodbc, psycopg2 o PyMySQL), sin autocommit: cada llamada a `atender` abre una, la confirma o la deshace y la cierra. Las tablas y columnas salen del CATÁLOGO, que el script de las tablas guarda en la base (XESQUEMA.CATALOGO, en la fila de la última versión). Este código lo lee de ahí y lo vuelve a leer cada minuto: cuando se ejecuta un script nuevo, las tablas y columnas nuevas se aceptan solas, sin reiniciar y sin cambiar este código. Probado con Python 3.11 (ver LEEME.txt). """ import json import logging import math import re import threading import time from datetime import datetime, timedelta, timezone from decimal import ROUND_HALF_UP, Decimal, InvalidOperation log = logging.getLogger('ramaflex') RESERVADOS = ('orden', 'pagina', 'limite') MAX_FILAS_PUT = 500 MAX_LIMITE = 1000 INT_MIN, INT_MAX = -2147483648, 2147483647 class Falla(Exception): """Error con su estado HTTP y su código de la especificación.""" def __init__(self, estado, codigo, mensaje): super().__init__(mensaje) self.estado = estado self.codigo = codigo def _invalido(mensaje): return Falla(400, 'DATO_INVALIDO', mensaje) def _paginar_limit(limite, salto): return ' LIMIT ? OFFSET ?', [limite, salto] # Lo único que cambia entre motores. El SQL se arma siempre con "?" y se traduce al marcador del # driver al ejecutarlo (psycopg2 y PyMySQL usan "%s"). # "conflicto": clave duplicada por dos escrituras simultáneas, y también el interbloqueo, que es # la forma en que suele terminar esa misma carrera (sobre todo en MySQL): ambos se pueden reintentar. MOTORES = { 'sqlserver': { 'marca': '?', # UPDLOCK + HOLDLOCK bloquea la clave aunque la fila aún no exista: dos PUT de la misma # clave nueva se esperan en vez de chocar. 'bloqueo_tabla': ' WITH (UPDLOCK, HOLDLOCK)', 'bloqueo_final': '', 'paginar': lambda limite, salto: (' OFFSET ? ROWS FETCH NEXT ? ROWS ONLY', [salto, limite]), # pyodbc no expone el número de error nativo: viene en el mensaje, "... (2627) (SQLExecDirectW)". 'conflicto': lambda e: re.search(r'\((2627|2601|1205)\) \(SQL', str(e)) is not None, }, 'postgresql': { 'marca': '%s', 'bloqueo_tabla': '', 'bloqueo_final': ' FOR UPDATE', 'paginar': _paginar_limit, 'conflicto': lambda e: getattr(e, 'pgcode', None) in ('23505', '40P01'), }, 'mysql': { 'marca': '%s', 'bloqueo_tabla': '', 'bloqueo_final': ' FOR UPDATE', 'paginar': _paginar_limit, 'conflicto': lambda e: bool(e.args) and e.args[0] in (1062, 1213), }, } # -------------------------------------------------------------------------------------------------- # Conversión de valores (pseudocódigo, sección 2) # -------------------------------------------------------------------------------------------------- _ENTERO = re.compile(r'-?[0-9]+') _DECIMAL = re.compile(r'-?[0-9]+(\.[0-9]+)?') # ISO-8601 con zona obligatoria. Se parsea a mano porque datetime.fromisoformat no acepta "Z" # ni fracciones de largo arbitrario antes de Python 3.11. _FECHA = re.compile(r'(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2})(?::(\d{2})(?:\.(\d+))?)?' r'(?:Z|([+-])(\d{2}):(\d{2}))', re.ASCII) def _fecha_a_utc(nombre, texto): """Texto ISO-8601 con zona -> datetime UTC sin zona (así se guarda), truncado a milisegundos.""" m = _FECHA.fullmatch(texto) if isinstance(texto, str) else None if not m: raise _invalido(f'{nombre}: se espera una fecha ISO-8601 con zona') a, mes, d, h, mi, s, frac, signo, zh, zm = m.groups() try: ms = int((frac or '0')[:3].ljust(3, '0')) utc = datetime(int(a), int(mes), int(d), int(h), int(mi), int(s or 0), ms * 1000) if signo: desfase = timedelta(hours=int(zh), minutes=int(zm)) utc -= desfase if signo == '+' else -desfase except (ValueError, OverflowError): raise _invalido(f'{nombre}: fecha inválida') return utc def _cabe(d, col): """¿Cabe d en DECIMAL(precision, escala)? A lo más (precision - escala) dígitos enteros, después de redondear a la escala como lo hará la base (9.999 en DECIMAL(3,2) pasa a 10.00 y no cabe).""" if 'precision' not in col: return True try: redondeado = d.quantize(Decimal(1).scaleb(-col['escala']), rounding=ROUND_HALF_UP) except InvalidOperation: # demasiados dígitos para redondear: no cabe return False return abs(redondeado) < Decimal(10) ** (col['precision'] - col['escala']) def convertir(nombre, col, valor, de_url=False): """Valor recibido (de JSON, o texto de la URL si de_url) -> valor para el driver.""" if valor is None: if not col['nulo']: raise _invalido(f'{nombre} no acepta null') return None tipo = col['tipo'] # En Python bool es un int: hay que descartarlo antes de aceptar números. es_numero = isinstance(valor, (int, float)) and not isinstance(valor, bool) if tipo == 'texto': if not (isinstance(valor, str) or es_numero): raise _invalido(f'{nombre}: se espera un texto') v = str(valor) if col.get('largo') is not None and len(v) > col['largo']: raise _invalido(f'{nombre}: más de {col["largo"]} caracteres') return v if tipo == 'entero': n = None if es_numero and (isinstance(valor, int) or valor.is_integer()): n = int(valor) elif isinstance(valor, str) and _ENTERO.fullmatch(valor): n = int(valor) if n is None or not INT_MIN <= n <= INT_MAX: # INT de 32 bits; fuera de rango la base daría 500 raise _invalido(f'{nombre}: se espera un entero entre {INT_MIN} y {INT_MAX}') return n if tipo == 'decimal': d = None if es_numero and (isinstance(valor, int) or math.isfinite(valor)): d = Decimal(str(valor)) elif isinstance(valor, str) and _DECIMAL.fullmatch(valor): d = Decimal(valor) if d is None or not _cabe(d, col): raise _invalido(f'{nombre}: se espera un número que quepa en la columna') return d if tipo == 'logico': if isinstance(valor, bool): return valor if de_url and valor in ('true', '1'): return True if de_url and valor in ('false', '0'): return False raise _invalido(f'{nombre}: se espera true o false') if tipo == 'fecha': return _fecha_a_utc(nombre, valor) raise _invalido(f'{nombre}: tipo desconocido en el catálogo') def _a_json(col, v): """Valor leído de la base -> valor JSON.""" if v is None: return None tipo = col['tipo'] if tipo == 'texto': return str(v).rstrip(' ') # los CHAR vienen rellenos con espacios if tipo == 'entero': return int(v) if tipo == 'decimal': return float(v) if tipo == 'logico': return bool(v) # MySQL devuelve 0/1 # fecha: el valor guardado ES UTC. Los tres drivers lo entregan sin zona y sin convertirlo; # si alguno lo entregara con zona, se lleva a UTC. if v.tzinfo is not None: v = v.astimezone(timezone.utc).replace(tzinfo=None) return v.isoformat(timespec='milliseconds') + 'Z' def _sin_constantes(nombre): raise ValueError(nombre) # json de Python acepta NaN e Infinity; JSON no # -------------------------------------------------------------------------------------------------- # Núcleo # -------------------------------------------------------------------------------------------------- class TablasRamaflex: def __init__(self, motor, conectar, refresco_segundos=60): """refresco_segundos: cada cuánto se vuelve a leer el catálogo de la base (60 s).""" if motor not in MOTORES: raise ValueError(f'Motor desconocido: {motor} (sqlserver, postgresql o mysql)') self.motor = MOTORES[motor] self.conectar = conectar self.refresco = refresco_segundos self._tablas = None # el catálogo leído ("tablas" del JSON) self._leido_en = 0.0 # time.monotonic() de la última lectura (o intento fallido con catálogo) self._candado = threading.Lock() # una sola lectura a la vez # ---- el catálogo, desde la base ----------------------------------------------------------------- def _vigente(self): tablas = self._tablas if tablas is not None and time.monotonic() - self._leido_en < self.refresco: return tablas return None def _definiciones(self): """Las tablas del catálogo; lo lee de XESQUEMA si nunca se leyó o si pasó el período.""" tablas = self._vigente() if tablas is not None: return tablas with self._candado: tablas = self._vigente() # otra llamada pudo leerlo mientras se esperaba el candado if tablas is not None: return tablas try: self._tablas = self._leer_catalogo() except Exception: if self._tablas is None: raise # Ya había uno: se sigue con él y se reintenta en el próximo período. log.warning('No se pudo releer el catálogo; se sigue con el anterior', exc_info=True) self._leido_en = time.monotonic() return self._tablas def _leer_catalogo(self): sql = 'SELECT CATALOGO FROM XESQUEMA WHERE VERSION = (SELECT MAX(VERSION) FROM XESQUEMA)' def trabajo(cur): self._ejecutar(cur, sql, []) fila = cur.fetchone() return fila[0] if fila else None try: texto = self._en_transaccion(trabajo) except Exception as e: raise RuntimeError('No se pudo leer XESQUEMA.CATALOGO (¿falta ejecutar el script actual de ' f'las tablas?): {e}') from e if not texto: raise RuntimeError('XESQUEMA no tiene el catálogo: falta ejecutar el script actual de las tablas.') return json.loads(texto)['tablas'] def atender(self, metodo, tabla, parametros, cuerpo_texto): """(metodo, tabla, [(nombre, valor), ...], texto del cuerpo) -> (estado HTTP, cuerpo JSON). Los nombres de tabla y columna recibidos se comparan EXACTO con el catálogo; como son iguales, el SQL lleva el nombre del catálogo. Los valores van siempre como parámetros. """ try: definicion = self._definiciones().get(tabla) if definicion is None: raise Falla(404, 'TABLA_DESCONOCIDA', f'Tabla desconocida: {tabla}') if metodo == 'GET': respuesta = self._listar(tabla, definicion, parametros) elif metodo in ('PUT', 'DELETE'): if definicion.get('soloLectura'): raise Falla(405, 'SOLO_LECTURA', f'{tabla} es de solo lectura.') if metodo == 'PUT': respuesta = self._guardar(tabla, definicion, cuerpo_texto) else: respuesta = self._borrar(tabla, definicion, parametros) else: raise Falla(405, 'METODO_NO_PERMITIDO', 'Solo GET, PUT y DELETE.') return 200, json.dumps(respuesta, ensure_ascii=False) except Falla as e: return e.estado, _error(e.codigo, str(e)) except Exception as e: if self.motor['conflicto'](e): return 409, _error('CONFLICTO', 'Otra escritura simultánea creó la misma clave; se puede reintentar.') # El detalle (SQL, datos) va solo al log del servidor, nunca a la respuesta. log.exception('Error al atender %s %s', metodo, tabla) return 500, _error('ERROR_INTERNO', 'Error interno.') # ---- operaciones ------------------------------------------------------------------------------ def _listar(self, tabla, d, parametros): where, valores = self._filtros(d, parametros) orden = [] for o in [v for n, v in parametros if n == 'orden']: for x in o.split(','): if not x: continue desc = x.startswith('-') col = x[1:] if desc else x _columna(d, col) if col not in [c for c, _ in orden]: # repetir una columna en ORDER BY es error en SQL Server orden.append((col, desc)) # Sin un orden total, las páginas se pisan: se completa con la clave. orden += [(c, False) for c in d['clave'] if c not in [x for x, _ in orden]] pagina = _entero_param(parametros, 'pagina', 1) limite = _entero_param(parametros, 'limite', MAX_LIMITE) if pagina < 1: raise _invalido('pagina debe ser 1 o más') if not 1 <= limite <= MAX_LIMITE: raise _invalido(f'limite debe estar entre 1 y {MAX_LIMITE}') pag_sql, pag_valores = self.motor['paginar'](limite, (pagina - 1) * limite) sql = (f'SELECT {", ".join(d["columnas"])} FROM {tabla}{where} ORDER BY ' + ', '.join(c + (' DESC' if desc else '') for c, desc in orden) + pag_sql) def trabajo(cur): self._ejecutar(cur, sql, valores + pag_valores) return [_fila_json(d, r) for r in cur.fetchall()] return {'filas': self._en_transaccion(trabajo)} def _guardar(self, tabla, d, cuerpo_texto): try: cuerpo = json.loads(cuerpo_texto or '', parse_constant=_sin_constantes) except ValueError: raise _invalido('El cuerpo no es un JSON válido.') filas = cuerpo.get('filas') if isinstance(cuerpo, dict) else None if not isinstance(filas, list) or len(filas) > MAX_FILAS_PUT: raise _invalido(f'Se espera {{ "filas": [...] }} con hasta {MAX_FILAS_PUT} filas.') # 1) Validar TODO antes de tocar la base. columnas, auto = d['columnas'], d.get('autonumerico') lote = [] for fila in filas: if not isinstance(fila, dict): raise _invalido('Cada fila debe ser un objeto.') for nombre in fila: _columna(d, nombre) if any(fila.get(c) is None for c in d['clave']): raise Falla(400, 'CLAVE_INCOMPLETA', 'Falta la clave: ' + ' + '.join(d['clave'])) # El autonumérico se ignora si viene: lo asigna la base y nunca se actualiza. lote.append({n: convertir(n, columnas[n], v) for n, v in fila.items() if n != auto}) obligatorias = [n for n, c in columnas.items() if not c['nulo'] and 'defecto' not in c and n != auto] where = ' WHERE ' + ' AND '.join(f'{c} = ?' for c in d['clave']) existe_sql = (f'SELECT 1 FROM {tabla}{self.motor["bloqueo_tabla"]}{where}' f'{self.motor["bloqueo_final"]}') leer_sql = f'SELECT {", ".join(columnas)} FROM {tabla}{where}' # 2) Escribir, todo o nada. def trabajo(cur): resultado = [] for v in lote: clave = [v[c] for c in d['clave']] self._ejecutar(cur, existe_sql, clave) if cur.fetchone(): resto = [c for c in v if c not in d['clave']] if resto: # solo lo enviado sets = ', '.join(f'{c} = ?' for c in resto) self._ejecutar(cur, f'UPDATE {tabla} SET {sets}{where}', [v[c] for c in resto] + clave) else: falta = [c for c in obligatorias if c not in v] if falta: raise _invalido('Falta ' + ', '.join(falta)) # Las que no vienen toman su DEFAULT o null: lo pone la base. marcas = ', '.join('?' for _ in v) self._ejecutar(cur, f'INSERT INTO {tabla} ({", ".join(v)}) VALUES ({marcas})', list(v.values())) self._ejecutar(cur, leer_sql, clave) resultado.append(_fila_json(d, cur.fetchone())) return resultado return {'filas': self._en_transaccion(trabajo)} def _borrar(self, tabla, d, parametros): where, valores = self._filtros(d, parametros) if not where: raise Falla(400, 'FILTRO_REQUERIDO', 'Borrar exige al menos un filtro.') def trabajo(cur): self._ejecutar(cur, f'DELETE FROM {tabla}{where}', valores) # Con SET NOCOUNT ON a nivel de servidor, SQL Server no informa el conteo (rowcount = -1). return cur.rowcount return {'borradas': self._en_transaccion(trabajo)} # ---- apoyo ------------------------------------------------------------------------------------ def _filtros(self, d, parametros): """Filtros de la query -> (' WHERE ...' o '', valores). Común a listar y borrar.""" condiciones, valores = [], [] nombres = [] for n, _ in parametros: if n not in RESERVADOS and n not in nombres: nombres.append(n) for nombre in nombres: todos = [v for n, v in parametros if n == nombre] base, _, sufijo = nombre.rpartition('.') if base and sufijo in ('desde', 'hasta'): col = _columna(d, base) valores.append(convertir(col, d['columnas'][col], todos[0], de_url=True)) condiciones.append(f'{col} {">=" if sufijo == "desde" else "<="} ?') # extremos incluidos else: col = _columna(d, nombre) xs = [convertir(col, d['columnas'][col], v, de_url=True) for v in todos] valores += xs if len(xs) == 1: condiciones.append(f'{col} = ?') else: condiciones.append(f'{col} IN ({", ".join("?" for _ in xs)})') return (' WHERE ' + ' AND '.join(condiciones)) if condiciones else '', valores def _ejecutar(self, cur, sql, valores): # El SQL solo tiene nombres del catálogo y palabras fijas: el único "?" es el marcador. cur.execute(sql.replace('?', self.motor['marca']), valores) def _en_transaccion(self, trabajo): con = self.conectar() try: resultado = trabajo(con.cursor()) con.commit() return resultado except Exception: try: con.rollback() except Exception: log.warning('No se pudo deshacer la transacción', exc_info=True) raise finally: con.close() def _columna(d, nombre): if nombre not in d['columnas']: raise Falla(400, 'COLUMNA_DESCONOCIDA', f'Columna desconocida: {nombre}') return nombre def _entero_param(parametros, nombre, defecto): valores = [v for n, v in parametros if n == nombre] if not valores: return defecto if not _ENTERO.fullmatch(valores[0]): raise _invalido(f'{nombre} debe ser un entero') return int(valores[0]) def _fila_json(d, registro): # Las columnas se leen por posición (el SELECT las pide en el orden del catálogo) y se # responden con el nombre del catálogo, aunque PostgreSQL las devuelva en minúsculas. return {n: _a_json(c, registro[i]) for i, (n, c) in enumerate(d['columnas'].items())} def _error(codigo, mensaje): return json.dumps({'ok': False, 'codigo': codigo, 'mensaje': mensaje}, ensure_ascii=False)