mtpk-mariadb 0.1.6

Librería para sincronización estructural de bases de datos MariaDB con modelos Python

mtpk_mariadb — Documentación Completa

ORM y wrapper asíncrono/síncrono para MariaDB, diseñado para proyectos Python con FastAPI y asyncio.

---

Índice

1. Instalación
2. Arquitectura del módulo
3. Clases de definición estructural
   • SQLLiteral
   • Columna
   • ForeignKey
   • Index
   • Tabla
4. Clases de conexión y consulta
   • Database (síncrona)
   • AsyncDatabase (asíncrona)
5. Sincronización de esquema
   • ManagerDB
6. CRUD asíncrono
   • AsyncCrudBase
7. Excepciones
   • MtpkErrorValidacionDb
   • MtpkErrorDb
8. Utilidades
   • Funciones de seguridad
   • Funciones SQL auxiliares
   • FiltroCampo
9. Interfaz unificada (interface.py)
10. Ejemplos completos

---

Instalación

pip install mtpk_mariadb

Dependencias principales: pymysql, aiomysql, bcrypt, pydantic.

---

Arquitectura del módulo

mtpk_mariadb/
├── core_sync.py        # Clases síncronas: Columna, Tabla, ForeignKey, Index, Database, ManagerDB
├── async_adapter.py    # AsyncDatabase — backend asíncrono con aiomysql
├── crud.py             # AsyncCrudBase — operaciones CRUD de alto nivel
├── interface.py        # Fachada unificada: exporta clases según modo sync/async
├── excepciones.py      # MtpkErrorDb, MtpkErrorValidacionDb
└── utils.py            # Utilidades: hashing, filtros SQL, generación de campos

Importación recomendada:

from mtpk_mariadb import Tabla, Columna, ForeignKey, Index, AsyncCrudBase
from mtpk_mariadb.async_adapter import AsyncDatabase
from mtpk_mariadb.excepciones import MtpkErrorDb, MtpkErrorValidacionDb
from mtpk_mariadb.utils import construir_condiciones_sql, generar_lista_campos

---

Clases de definición estructural

SQLLiteral

Permite insertar un valor literal en SQL (sin comillas), útil para valores por defecto como CURRENT_TIMESTAMP o expresiones SQL.

from mtpk_mariadb.core_sync import SQLLiteral

# Uso en Columna.default para evitar que el valor sea entrecomillado
Columna(nombre="fecha_crea", tipo="DATETIME", default=SQLLiteral("CURRENT_TIMESTAMP"))

Atributos:

• valor (str): Texto literal que se inyecta directamente en el SQL generado.

---

Columna

Representa una columna de una tabla MariaDB. Se usa como @dataclass.

Atributos:

Atributo	Tipo	Por defecto	Descripción
nombre	str	—	Nombre de la columna (obligatorio)
tipo	Literal[...]	—	Tipo SQL: INT, VARCHAR, DECIMAL, TEXT, DATETIME, BOOLEAN, ENUM, SET, BLOB, etc.
longitud	Optional[int]	None	Longitud para VARCHAR(n), CHAR(n), BINARY(n), VARBINARY(n)
precision	Optional[int]	None	Precisión para DECIMAL(p,s), FLOAT(p)
escala	Optional[int]	None	Escala para DECIMAL(p,s)
not_null	bool	False	Si la columna es NOT NULL
primary_key	bool	False	Si es clave primaria. Implica NOT NULL automáticamente
unique	bool	False	Restricción UNIQUE
auto_increment	bool	False	AUTO_INCREMENT (solo para enteros con PK)
default	str|int|float|SQLLiteral	None	Valor por defecto. Usar SQLLiteral para expresiones SQL
comentario	Optional[str]	None	COMMENT de la columna en MariaDB
enum_opciones	Optional[list[str]]	None	Valores válidos cuando tipo es ENUM o SET
protegido_insertar	bool	False	Si True, la columna se ignora en insert() cuando aplicar_protegidos=True
protegido_actualizar	bool	False	Si True, la columna se ignora en update() cuando aplicar_protegidos=True
generado	Optional[str]	None	Expresión SQL para columnas calculadas: GENERATED ALWAYS AS (expresion)
generado_tipo	Optional[Literal["STORED", "VIRTUAL"]]	None	Modo de la columna generada: STORED (persiste en disco) o VIRTUAL (calcula al leer)
tabla	Optional[Tabla]	None	Referencia interna a la tabla padre. Se asigna automáticamente

Métodos:

• to_sql() -> str: Genera la definición SQL completa de la columna.

Ejemplos:

from mtpk_mariadb import Columna
from mtpk_mariadb.core_sync import SQLLiteral

# Clave primaria autoincremental
Columna(nombre="id", tipo="INT", auto_increment=True, primary_key=True, not_null=True)

# VARCHAR con longitud
Columna(nombre="nombre", tipo="VARCHAR", longitud=200, not_null=True)

# DECIMAL con precisión y escala
Columna(nombre="importe", tipo="DECIMAL", precision=12, escala=2, not_null=True, default=0)

# DATETIME con valor por defecto SQL
Columna(nombre="fecha_crea", tipo="DATETIME", default=SQLLiteral("CURRENT_TIMESTAMP"))

# ENUM
Columna(nombre="estado", tipo="ENUM", enum_opciones=["activo", "inactivo", "borrador"], default="activo")

# Columna calculada (GENERATED STORED)
Columna(
    nombre="importe_total",
    tipo="DECIMAL", precision=12, escala=2,
    generado="ROUND(base_imponible + importe_iva, 2)",
    generado_tipo="STORED"
)

# Columna protegida (no se modifica por CRUD genérico)
Columna(nombre="fecha_crea", tipo="DATETIME", protegido_actualizar=True)

---

ForeignKey

Representa una clave foránea (FOREIGN KEY) entre tablas. Se usa como @dataclass.

Atributos:

Atributo	Tipo	Por defecto	Descripción
columna	str	—	Nombre de la columna local que actúa como FK (obligatorio)
referencia_tabla	str	—	Tabla referenciada (obligatorio)
referencia_columna	str	—	Columna de la tabla referenciada (obligatorio)
nombre	Optional[str]	None	Nombre explícito del constraint. Si no se indica, MariaDB lo genera
on_delete	Optional[str]	None	Acción al eliminar: CASCADE, RESTRICT, SET NULL, NO ACTION
on_update	Optional[str]	None	Acción al actualizar: CASCADE, RESTRICT, SET NULL, NO ACTION
tabla	Optional[Tabla]	None	Referencia interna a la tabla padre. Se asigna automáticamente

Métodos:

• to_sql() -> str: Genera la cláusula SQL del constraint.

Ejemplo:

from mtpk_mariadb import ForeignKey

ForeignKey(
    columna="id_cliente",
    referencia_tabla="clientes",
    referencia_columna="id",
    nombre="fk_pedidos_clientes",
    on_delete="RESTRICT",
    on_update="CASCADE"
)

---

Index

Representa un índice (simple o compuesto, normal o único) en una tabla. Se usa como @dataclass.

Atributos:

Atributo	Tipo	Por defecto	Descripción
columnas	List[str]	—	Lista de columnas que componen el índice (obligatorio)
nombre	Optional[str]	None	Nombre del índice. Si no se indica, se genera automáticamente
unico	bool	False	Si True, genera UNIQUE KEY; si False, genera KEY normal
tabla	Optional[Tabla]	None	Referencia interna a la tabla padre. Se asigna automáticamente

Métodos:

• to_sql() -> str: Genera la cláusula SQL del índice.

Ejemplos:

from mtpk_mariadb import Index

# Índice simple
Index(columnas=["id_cliente"])

# Índice compuesto
Index(columnas=["id_obligado", "ejercicio"], nombre="idx_obligado_ejercicio")

# Índice único
Index(columnas=["email"], unico=True)

---

Tabla

Representa la definición completa de una tabla MariaDB. Se usa como @dataclass.

Atributos:

Atributo	Tipo	Por defecto	Descripción
nombre	str	—	Nombre de la tabla (obligatorio)
columnas	List[Columna]	—	Lista de columnas (obligatorio)
indices	List[Index]	[]	Índices adicionales
foreign_keys	List[ForeignKey]	[]	Claves foráneas
claves_primarias	List[str]	[]	PKs compuestas (si no están en Columna.primary_key)
comentario	Optional[str]	None	Comentario de tabla en MariaDB
engine	str	"InnoDB"	Motor de almacenamiento
charset	str	"utf8mb4"	Charset de la tabla
registros_iniciales	Optional[List[dict]]	[]	Registros a insertar automáticamente al crear la tabla
vistas	Optional[List[str]]	[]	Sentencias SQL de vistas asociadas
triggers	Optional[list[str]]	None	Sentencias SQL de triggers
procedimientos	Optional[list[str]]	None	Sentencias SQL de procedimientos almacenados
db_name	Optional[str]	None	Nombre de la base de datos a la que pertenece (clasificación)
database	Optional[Database]	None	Referencia a la instancia Database. Se asigna automáticamente con add_tabla()

Métodos:

Método	Descripción
add_columna(columna)	Añade una columna. Lanza ValueError si ya existe una con el mismo nombre
set_columnas(columnas)	Reemplaza todas las columnas. Valida duplicados
get_nombres_columnas() -> list[str]	Devuelve los nombres de todas las columnas
add_index(index)	Añade un índice. Lanza ValueError si ya existe uno con las mismas columnas
set_indices(indices)	Reemplaza todos los índices. Valida duplicados
add_foreign_key(fk)	Añade una FK. Lanza ValueError si ya existe una equivalente
set_foreign_keys(foreign_keys)	Reemplaza todas las FKs. Valida duplicados
set_database(db)	Establece la referencia a la Database padre
to_sql() -> str	Genera la sentencia CREATE TABLE IF NOT EXISTS completa
comparar_generar_alter(conexion, dbname, permitir_drop) -> List[str]	Compara con la BD real y genera sentencias ALTER TABLE necesarias
obtener_columnas_mariadb(conexion, dbname) -> List[Columna]	Introspección: obtiene columnas reales desde INFORMATION_SCHEMA
obtener_indices_mariadb(conexion, dbname) -> List[Index]	Introspección: obtiene índices reales desde SHOW INDEX
obtener_fks_mariadb(conexion, dbname) -> List[ForeignKey]	Introspección: obtiene FKs reales desde information_schema
extraer_estructura(db_name) -> list[tuple]	Devuelve la estructura como lista de tuplas (db_name, tabla, tipo, nombre, definicion)
extraer_longitud_campo(tipo_sql) -> Optional[int]	Extrae la longitud de un tipo SQL como VARCHAR(255) → 255

Ejemplo:

from mtpk_mariadb import Tabla, Columna, ForeignKey, Index
from mtpk_mariadb.core_sync import SQLLiteral

TABLA_PEDIDOS = Tabla(
    nombre="pedidos",
    comentario="Pedidos de venta",
    columnas=[
        Columna(nombre="id", tipo="INT", auto_increment=True, primary_key=True),
        Columna(nombre="id_cliente", tipo="INT", not_null=True),
        Columna(nombre="fecha", tipo="DATE", not_null=True),
        Columna(nombre="total", tipo="DECIMAL", precision=12, escala=2, default=0),
        Columna(nombre="estado", tipo="ENUM", enum_opciones=["pendiente", "enviado", "cancelado"], default="pendiente"),
        Columna(nombre="fecha_crea", tipo="DATETIME", default=SQLLiteral("CURRENT_TIMESTAMP"), protegido_actualizar=True),
    ],
    indices=[
        Index(columnas=["id_cliente"]),
        Index(columnas=["fecha", "estado"], nombre="idx_fecha_estado"),
    ],
    foreign_keys=[
        ForeignKey(columna="id_cliente", referencia_tabla="clientes", referencia_columna="id", on_delete="RESTRICT"),
    ]
)

---

Clases de conexión y consulta

Database (síncrona)

Gestiona conexiones síncronas con MariaDB usando pymysql. Base de ManagerDB.

Constructor:

Database(host, user, password, db, port=3306, logger=None)

Atributos públicos:

Atributo	Tipo	Descripción
host	str	Host de la BD
user	str	Usuario
password	str	Contraseña
db	str	Nombre de la base de datos
port	int	Puerto (default 3306)
tablas	Dict[str, Tabla]	Tablas registradas
ultimo_insert_id	Optional[int]	ID del último INSERT ejecutado

Métodos:

Método	Descripción
conectar(autocommit=False)	Abre la conexión reutilizable. Devuelve la conexión pymysql
cerrar()	Cierra la conexión si está abierta
commit()	Confirma la transacción activa y cierra la conexión
rollback()	Revierte la transacción activa y cierra la conexión
transaccion()	Context manager síncrono para ejecutar operaciones en una transacción
add_tabla(tabla)	Registra una Tabla en la base de datos
get_tabla(nombre) -> Tabla	Devuelve la Tabla registrada por nombre. Lanza KeyError si no existe
query(sql, params, conexion, uno) -> Any	Detecta automáticamente SELECT o acción. Devuelve lista, dict o nº de filas
call_proc(nombre, parametros, conexion, uno)	Ejecuta un procedimiento almacenado. Devuelve dict o list[dict]

Detección automática en query():

• Empieza por SELECT, SHOW, DESC, DESCRIBE, EXPLAIN → delega en _query_select
• Cualquier otro comando → delega en _query_accion (INSERT, UPDATE, DELETE...)

Ejemplo:

from mtpk_mariadb.core_sync import Database

db = Database(host="localhost", user="root", password="1234", db="mi_bd")

# Consulta simple
filas = db.query("SELECT * FROM pedidos WHERE estado = %s", ("pendiente",))

# Inserción
db.query("INSERT INTO pedidos (id_cliente, total) VALUES (%s, %s)", (1, 150.00))
nuevo_id = db.ultimo_insert_id

# Transacción manual
conn = db.conectar()
try:
    db.query("UPDATE stock SET cantidad = cantidad - 1 WHERE id = %s", (5,), conexion=conn)
    db.query("INSERT INTO movimientos (id_producto) VALUES (%s)", (5,), conexion=conn)
    db.commit()
except Exception:
    db.rollback()

# Context manager de transacción
conn = db.conectar()
with db.transaccion() as conn:
    db.query("UPDATE ...", conexion=conn)
    db.query("INSERT ...", conexion=conn)

---

AsyncDatabase (asíncrona)

Backend asíncrono con aiomysql. Diseñado para entornos FastAPI/Uvicorn. Internamente usa _PseudoPool: cada operación abre y cierra su propia conexión (sin reutilización real de pool), lo que simplifica la gestión en contextos concurrentes.

Constructor:

AsyncDatabase(host, user, password, db, port=3306, logger=None)

Atributos públicos:

Atributo	Tipo	Descripción
host, user, password, db, port	str/int	Parámetros de conexión
tablas	Dict[str, Tabla]	Tablas registradas
ultimo_insert_id	Optional[int]	ID del último INSERT
pool	_PseudoPool|None	Pool interno (se crea automáticamente en la primera operación)

Métodos:

Método	Async	Descripción
conectar()	async	Inicializa el pool interno si no existe. Se llama automáticamente
cerrar()	async	No-op seguro. Mantiene compatibilidad con código legacy
en_transaccion() -> bool	No	Indica si hay una transacción abierta en la coroutine actual
transaccion(autocommit=False)	async context manager	Abre conexión, hace BEGIN, cede, COMMIT/ROLLBACK automático
transaccion_simple	alias	Alias de transaccion() por compatibilidad
add_tabla(tabla)	No	Registra una Tabla
get_tabla(nombre) -> Tabla	No	Devuelve una Tabla registrada. Lanza KeyError si no existe
query(sql, params, conexion, uno)	async	Detecta SELECT/acción automáticamente
query_multi_action(sql, lista_valores, conexion)	async	Ejecuta executemany para múltiples valores con la misma sentencia
call_proc(nombre, parametros, conexion, uno)	async	Ejecuta un procedimiento almacenado

Conversores automáticos:

Los tipos DATE, DATETIME, TIMESTAMP se devuelven como str. DECIMAL y NEWDECIMAL como Decimal.

Ejemplo:

from mtpk_mariadb.async_adapter import AsyncDatabase

config = {"host": "localhost", "user": "root", "password": "1234", "db": "mi_bd"}

async def ejemplo():
    db = AsyncDatabase(**config)

    # Consulta simple (gestión de conexión automática)
    filas = await db.query("SELECT * FROM pedidos")

    # Una sola fila
    pedido = await db.query("SELECT * FROM pedidos WHERE id = %s", (42,), uno=True)

    # Transacción
    async with db.transaccion() as conn:
        await db.query("UPDATE stock SET cantidad = cantidad - 1 WHERE id = %s", (5,), conexion=conn)
        await db.query("INSERT INTO movimientos (id_producto) VALUES (%s)", (5,), conexion=conn)
    # El commit se aplica automáticamente al salir del bloque

    # Múltiples inserciones
    datos = [(1, 10.0), (2, 20.0), (3, 30.0)]
    await db.query_multi_action(
        "INSERT INTO lineas (id_pedido, importe) VALUES (%s, %s)",
        datos
    )

---

Sincronización de esquema

ManagerDB

Extiende Database con funcionalidades avanzadas para crear y sincronizar el esquema de base de datos. Usa _migraciones y estructura_actual como tablas de control.

Constructor: idéntico a Database.

Métodos:

Método	Descripción
crear_tablas_si_no_existen() -> dict	Ejecuta CREATE TABLE IF NOT EXISTS para todas las tablas registradas. No aplica ALTER
simular_cambios(permitir_drop=False) -> dict	Genera pero NO ejecuta las sentencias SQL necesarias. Devuelve {tabla: [sentencias]}
simular_cambios_avanzado() -> dict	Simula comparando tanto contra la BD real como contra estructura_actual
aplicar_cambios(permitir_drop=True) -> dict	Aplica CREATE y ALTER en una transacción. Registra cada cambio en _migraciones
aplicar_cambios_respecto_a_estructura_actual(permitir_drop=False) -> dict	Aplica cambios comparando contra la estructura previamente registrada en estructura_actual. En primera ejecución equivale a aplicar_cambios()
verificar_contra_estructura_actual() -> dict	Detecta diferencias entre los modelos actuales y la estructura guardada. No aplica cambios
guardar_estructura_actual(conn)	Persiste la estructura actual de los modelos en la tabla estructura_actual
crear_tabla_estructura_si_no_existe(conn)	Crea la tabla estructura_actual si no existe

Tabla _migraciones (gestionada automáticamente):

CREATE TABLE IF NOT EXISTS _migraciones (
    id INT AUTO_INCREMENT PRIMARY KEY,
    tabla VARCHAR(100),
    sentencia TEXT,
    fecha DATETIME DEFAULT CURRENT_TIMESTAMP
)

Tabla estructura_actual (gestionada automáticamente):

CREATE TABLE IF NOT EXISTS estructura_actual (
    id INT AUTO_INCREMENT PRIMARY KEY,
    db_name VARCHAR(100) NOT NULL,
    tabla VARCHAR(100) NOT NULL,
    tipo ENUM('tabla', 'columna', 'indice', 'fk') NOT NULL,
    nombre VARCHAR(100),
    definicion TEXT NOT NULL,
    fecha_actualizacion DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uniq_estructura (db_name, tabla, tipo, nombre)
)

Flujo de uso recomendado:

from mtpk_mariadb.core_sync import ManagerDB

manager = ManagerDB(host="localhost", user="root", password="1234", db="mi_bd")
manager.add_tabla(TABLA_CLIENTES)
manager.add_tabla(TABLA_PEDIDOS)

# Simular antes de aplicar
cambios = manager.simular_cambios()
for tabla, sqls in cambios.items():
    for sql in sqls:
        print(f"[{tabla}] {sql}")

# Aplicar cambios reales
manager.aplicar_cambios(permitir_drop=False)

Gestión de dependencias entre tablas:

ManagerDB ordena automáticamente las tablas según sus claves foráneas (orden topológico) antes de crear o alterar, evitando errores por referencias a tablas que aún no existen.

---

CRUD asíncrono

AsyncCrudBase

Capa de acceso a datos de alto nivel, diseñada para ser heredada por los modelos de la aplicación. Cada instancia está asociada a una Tabla específica.

Constructor:

AsyncCrudBase(tabla: Tabla, config: dict | Callable[[], dict], logger=None)

• config puede ser un dict estático o un Callable que devuelve el dict (útil para config dinámica).

Métodos:

insert(datos, db, conexion) -> int

Inserta un registro. Solo se insertan columnas que existen en la definición de la tabla (ignora campos extra automáticamente).

nuevo_id = await crud.insert({"nombre": "Ana", "email": "[e-mail]"})

update(id, datos, db, conexion) -> int

Actualiza un registro por su id. Solo actualiza columnas válidas de la tabla.

filas_afectadas = await crud.update(42, {"nombre": "Ana García"})

eliminar(id, db, conexion) -> int

Elimina un registro por su id.

filas = await crud.eliminar(42)

obtener(id, alias, db, conexion) -> Optional[dict]

Obtiene un registro por su id. Devuelve None si no existe.

registro = await crud.obtener(42)

listar(offset, limit, orden, filtros, alias, db, conexion) -> Tuple[list[dict], int]

Lista registros con paginación, ordenación y filtros. Devuelve (lista, total).

• Si el número de resultados es igual a limit, realiza automáticamente una segunda consulta COUNT(*) para obtener el total real.
• El parámetro alias permite mapear nombres de campo a expresiones SQL (útil en JOINs).

filas, total = await crud.listar(
    offset=0,
    limit=50,
    orden="nombre ASC",
    filtros={
        "estado": {"op": "=", "valor": "activo"},
        "fecha_crea": {"op": "between", "valor": {"desde": "2025-01-01", "hasta": "2025-12-31"}},
        "nombre": {"op": "like", "valor": "%garcia%"},
    }
)

_normalizar_valores(obj) -> Any

Normaliza recursivamente cualquier valor Python a tipos aceptados por la BD:

Tipo entrada	Tipo salida
None, str, int, float, bool	Sin cambio
dict / Mapping	dict normalizado recursivamente
list, tuple, set, frozenset	list normalizado recursivamente
Enum	enum.value
UUID, Decimal, Path	str
date, datetime	ISO format str
dataclass	dict via asdict(), luego normalizado
Pydantic v2 (model_dump)	dict normalizado
bytes, bytearray, memoryview	bytes (para BLOB)

Puedes extender los conversores añadiendo entradas a NORMALIZADORES_EXTRA:

class MiCrud(AsyncCrudBase):
    NORMALIZADORES_EXTRA = {
        MiClaseCustom: lambda x: x.to_string()
    }

Parámetro db y conexion:

Ambos métodos CRUD aceptan db (instancia AsyncDatabase) y conexion (conexión activa de una transacción). Si se pasan, se reutilizan; si no, se crean internamente. Esto permite composición dentro de una misma transacción:

async with db.transaccion() as conn:
    id_pedido = await crud_pedidos.insert(datos_pedido, db=db, conexion=conn)
    for linea in lineas:
        linea["id_pedido"] = id_pedido
        await crud_lineas.insert(linea, db=db, conexion=conn)
# Commit automático al salir

Métodos obsoletos (mantienen compatibilidad):

• insertar(datos, aplicar_protegidos, db, conexion) → usa insert() en su lugar.
• actualizar(id, datos, aplicar_protegidos, db, conexion) → usa update() en su lugar.

---

Excepciones

MtpkErrorValidacionDb

Para errores de validación lógica (restricciones, formatos inválidos, consistencia de datos).

from mtpk_mariadb.excepciones import MtpkErrorValidacionDb

raise MtpkErrorValidacionDb("El NIF introducido no es válido", codigo="VAL-001")

Atributos: mensaje (str), codigo (str).

MtpkErrorDb

Para errores críticos de base de datos. Registra automáticamente el error y la traza si se pasa un logger.

from mtpk_mariadb.excepciones import MtpkErrorDb

raise MtpkErrorDb("Error al insertar factura", codigo="DB-500", logger=self.logger)

Atributos: mensaje (str), codigo (str).

Patrón de captura recomendado en FastAPI:

except HTTPException:
    raise
except MtpkErrorValidacionDb as exc:
    raise HTTPException(status_code=422, detail=exc.mensaje)
except MtpkErrorDb as exc:
    raise HTTPException(status_code=409, detail=exc.mensaje)
except Exception as exc:
    raise HTTPException(status_code=500, detail="Error interno")

---

Utilidades

Funciones de seguridad

from mtpk_mariadb.utils import (
    hash_password,
    verify_password,
    generar_api_key,
    generar_api_secret,
    verificar_api_secret,
)

hash_password(plain_password: str) -> str

Genera un hash bcrypt seguro. El resultado está listo para almacenar en la BD.

hash = hash_password("miContraseña123")

verify_password(plain_password: str, hashed_password: str) -> bool

Verifica si una contraseña en texto plano coincide con su hash bcrypt.

ok = verify_password("miContraseña123", hash_almacenado)

generar_api_key() -> str

Genera una API key aleatoria de ~43 caracteres URL-safe.

api_key = generar_api_key()  # "xK9mP2..."

generar_api_secret() -> tuple[str, str]

Genera un API secret aleatorio de ~65 caracteres. Devuelve (secreto_plano, secreto_hash_sha256). Almacenar solo el hash.

secreto, secreto_hash = generar_api_secret()
# Guardar en BD: secreto_hash
# Entregar al usuario: secreto

verificar_api_secret(api_secret: str, hashed_api_secret: str) -> bool

Verifica un API secret usando comparación de tiempo constante (segura frente a ataques de timing).

ok = verificar_api_secret(secreto_recibido, hash_almacenado)

---

Funciones SQL auxiliares

from mtpk_mariadb.utils import (
    generar_lista_campos,
    construir_condiciones_sql,
    resolver_orden_sql,
)

generar_lista_campos(campos, alias=None, excluir=None, prefijo_alias=None) -> str

Genera una lista de campos SQL a partir de un dict, lista de strings o lista de Columna. Útil para construir SELECT con alias.

# Con alias de tabla
campos_sql = generar_lista_campos(
    TABLA_PEDIDOS.columnas,
    alias="p",
    excluir=["password"],
    prefijo_alias="pedido_"
)
# -> "p.id AS pedido_id, p.nombre AS pedido_nombre, ..."

Parámetro	Descripción
campos	dict, list[str] o list[Columna]
alias	Alias de tabla SQL (ej: "p" → p.campo)
excluir	Lista de nombres de campos a omitir
prefijo_alias	Prefijo para el alias de salida (ej: "p_" → p.campo AS p_campo)

construir_condiciones_sql(filtros, alias=None) -> tuple[str, list]

Construye la cláusula WHERE (sin el WHERE) a partir de un diccionario de filtros estructurado.

Cada entrada del dict sigue el formato: {campo: {"op": operador, "valor": valor}}.

Operadores soportados:

Operador	Comportamiento
"="	Igual. Si valor es None → IS NULL. Si es lista → IN (...)
"!="	Distinto. Si valor es None → IS NOT NULL. Si es lista → NOT IN (...)
">", "<", ">=", "<="	Comparación numérica/fecha
"like"	LIKE %s
"in"	IN (...) para lista de valores
"between"	BETWEEN desde AND hasta. Acepta lista [desde, hasta] o dict {"desde": x, "hasta": y}

condiciones, valores = construir_condiciones_sql({
    "estado": {"op": "=", "valor": "activo"},
    "id_cliente": {"op": "in", "valor": [1, 2, 3]},
    "fecha": {"op": "between", "valor": {"desde": "2025-01-01", "hasta": "2025-12-31"}},
    "nombre": {"op": "like", "valor": "%garcia%"},
    "eliminado": {"op": "=", "valor": None},  # -> IS NULL
}, alias={"fecha": "p.fecha_pedido"})

sql = f"SELECT * FROM pedidos WHERE {condiciones}"

resolver_orden_sql(orden: str, alias=None) -> str

Convierte una cadena de ordenación a SQL seguro, aplicando alias si se definen.

orden_sql = resolver_orden_sql("nombre DESC, fecha ASC", alias={"nombre": "p.nombre_cliente"})
# -> "p.nombre_cliente DESC, fecha ASC"

---

FiltroCampo

Modelo Pydantic para validar filtros recibidos en endpoints de la API.

from mtpk_mariadb.utils import FiltroCampo

class FiltrosBusqueda(BaseModel):
    estado: Optional[FiltroCampo] = None
    fecha: Optional[FiltroCampo] = None

# Ejemplo de uso desde un endpoint:
filtros = {}
if data.estado:
    filtros["estado"] = {"op": data.estado.op, "valor": data.estado.valor}

Atributos:

• op: Operador SQL. Uno de: "=", "!=", ">", "<", ">=", "<=", "like", "in", "between"
• valor: El valor a comparar (cualquier tipo).

---

Interfaz unificada (interface.py)

Permite seleccionar automáticamente el backend síncrono o asíncrono.

from mtpk_mariadb.interface import set_async_mode, is_async_context, resolve_backend

Función	Descripción
set_async_mode(value: bool)	Fuerza el modo async globalmente (útil al arrancar FastAPI)
is_async_context() -> bool	Detecta si hay un loop asyncio activo en el hilo actual
resolve_backend(async_mode=None)	Devuelve el módulo backend correcto (core_sync o async_adapter)

Las clases Tabla, Columna, ForeignKey, Index, Database exportadas por mtpk_mariadb se resuelven automáticamente contra el backend activo.

---

Ejemplos completos

Definir un modelo con su tabla y CRUD

# database/estructura/modelos/_mod_clientes.py
from mtpk_mariadb import Tabla, Columna, Index, AsyncCrudBase
from mtpk_mariadb.core_sync import SQLLiteral

TABLA_CLIENTES = Tabla(
    nombre="clientes",
    comentario="Tabla de clientes",
    columnas=[
        Columna(nombre="id", tipo="INT", auto_increment=True, primary_key=True),
        Columna(nombre="id_obligado", tipo="INT", not_null=True),
        Columna(nombre="nombre", tipo="VARCHAR", longitud=200, not_null=True),
        Columna(nombre="nif", tipo="VARCHAR", longitud=20),
        Columna(nombre="email", tipo="VARCHAR", longitud=150),
        Columna(nombre="activo", tipo="TINYINT", not_null=True, default=1),
        Columna(nombre="fecha_crea", tipo="DATETIME", default=SQLLiteral("CURRENT_TIMESTAMP"), protegido_actualizar=True),
    ],
    indices=[
        Index(columnas=["id_obligado"]),
        Index(columnas=["nif"], unico=True),
    ]
)

def get_config():
    from config import CONFIG_DB
    return CONFIG_DB

class ModeloClientes(AsyncCrudBase):
    def __init__(self, logger=None):
        super().__init__(TABLA_CLIENTES, config=get_config, logger=logger)

    async def crear(self, id_obligado: int, nombre: str, nif: str = None, email: str = None) -> int:
        datos = self._normalizar_valores({
            "id_obligado": id_obligado,
            "nombre": nombre,
            "nif": nif,
            "email": email,
        })
        return await self.insert(datos)

    async def obtener_por_nif(self, nif: str, id_obligado: int):
        from mtpk_mariadb.async_adapter import AsyncDatabase
        db = AsyncDatabase(**get_config())
        return await db.query(
            "SELECT * FROM clientes WHERE nif = %s AND id_obligado = %s",
            (nif, id_obligado), uno=True
        )

Sincronización de esquema al arrancar

# sync.py
from mtpk_mariadb.core_sync import ManagerDB
from database.estructura.dependencias._dep_clientes import TABLA_CLIENTES
from database.estructura.dependencias._dep_pedidos import TABLA_PEDIDOS

def sincronizar_db():
    manager = ManagerDB(
        host="localhost", user="app_user", password="secret", db="mi_bd",
        logger=logging.getLogger("sync")
    )
    manager.add_tabla(TABLA_CLIENTES)
    manager.add_tabla(TABLA_PEDIDOS)

    cambios = manager.simular_cambios()
    for tabla, sqls in cambios.items():
        for sql in sqls:
            print(f"[PENDIENTE] {tabla}: {sql}")

    manager.aplicar_cambios(permitir_drop=False)

Transacción multi-tabla desde un endpoint FastAPI

# apps/api/v1/metodos/_api_pedidos.py
from mtpk_mariadb.async_adapter import AsyncDatabase
from mtpk_mariadb.excepciones import MtpkErrorDb, MtpkErrorValidacionDb

async def crear_pedido_con_lineas(datos_pedido, lineas, config):
    db = AsyncDatabase(**config)

    async with db.transaccion() as conn:
        # Insertar cabecera
        id_pedido = await crud_pedidos.insert(datos_pedido, db=db, conexion=conn)

        # Insertar líneas
        for linea in lineas:
            linea["id_pedido"] = id_pedido
            await crud_lineas.insert(linea, db=db, conexion=conn)

    # El commit se aplica automáticamente al salir del bloque sin excepción
    return id_pedido

Uso de filtros dinámicos en listar

filas, total = await modelo.listar(
    offset=0,
    limit=100,
    orden="fecha_crea DESC",
    filtros={
        "id_obligado": {"op": "=", "valor": 1},
        "activo": {"op": "=", "valor": 1},
        "nombre": {"op": "like", "valor": "%garcia%"},
        "fecha_crea": {
            "op": "between",
            "valor": {"desde": "2025-01-01", "hasta": "2025-12-31"}
        },
    }
)

---

Notas importantes

• Tipos de retorno de fechas: DATE, DATETIME y TIMESTAMP se devuelven siempre como str (formato ISO). No como objetos datetime.
• Tipos de retorno numéricos: DECIMAL se devuelve como Decimal, no como float.
• Columnas generadas: Los campos GENERATED ALWAYS AS no deben incluirse en los datos de insert() ni update(). MariaDB los calcula automáticamente.
• protegido_insertar / protegido_actualizar: Solo tienen efecto cuando se usa el método obsoleto insertar(aplicar_protegidos=True). Los métodos insert() y update() modernos no usan estas marcas.
• Seguridad multi-tenant: El campo id_obligado debe validarse siempre a nivel de aplicación antes de llegar al modelo. La librería no impone restricciones de tenant.