SQLAchemy-iris con la última versión del controlador Python

Después de tantos años de espera, por fin tenemos un controlador oficial disponible en Pypi

Además, encontré el driver JDBC finalmente disponible en Maven desde hace ya 3 meses, y el driver .Net en Nuget desde hace más de un mes.

Como autor de multitud de implementaciones de librerías de Python con soporte de IRIS, quería probarlo. La implementación de DB-API significa que debe ser reemplazable e implementar las funciones definidas en el estándar. La única diferencia entre las diferentes bases de datos debería ser únicamente el SQL.

Y la belleza de usar librerías existentes que ya implementaron la conectividad con otras bases de datos usando el estándar DB-API, es que estas librerías deberían ser totalmente compatibles con el controlador.

Decidí probar el driver oficial de InterSystems implementando su soporte en la librería SQLAlchemy-iris.

executemany

Prepara una operación a la base de datos (ya sea consulta o comando) y los ejecuta contra toda la lista de parámetros o mapeos encontrados en la variable de entrada seq_of_parameters.

Esta función es muy util para insertar varias filas a la vez. Empecemos con un ejemplo sencillo

import iris

host = "localhost"
port = 1972
namespace = "USER"
username = "_SYSTEM"
password = "SYS"
conn = iris.connect(
    host,
    port,
    namespace,
    username,
    password,
)

with conn.cursor() as cursor:
    cursor = conn.cursor()

    res = cursor.execute("DROP TABLE IF EXISTS test")
    res = cursor.execute(
        """
    CREATE TABLE test (
            id IDENTITY NOT NULL,
            value VARCHAR(50)
    ) WITH %CLASSPARAMETER ALLOWIDENTITYINSERT = 1
    """
    )

    cursor = conn.cursor()
    res = cursor.executemany(
        "INSERT INTO test (id, value) VALUES (?, ?)", [
            (1, 'val1'),
            (2, 'val2'),
            (3, 'val3'),
            (4, 'val4'),
        ]
    )

Esto funciona bien, pero ¿qué pasa si necesitamos insertar sólo un valor por cada fila?

    res = cursor.executemany(
        "INSERT INTO test (value) VALUES (?)", [
            ('val1', ),
            ('val2', ),
            ('val3', ),
            ('val4', ),
        ]
    )

Lamentablemente, esto da lugar a una excepción inesperada

RuntimeError: Cannot use list/tuple for single values

Por alguna razón, se permite un valor por fila pero InterSystems requiere una forma distinta de enviar los valores:

    res = cursor.executemany(
        "INSERT INTO test (value) VALUES (?)", [
            'val1',
            'val2',
            'val3',
            'val4',
        ]
    )

Así funciona bien

fetchone

Obiente la siguiente fila del conjunto de resultados de una consulta, devolviendo una única secuencia o ninguna cuando no hay más datos disponibles.

Por ejemplo, un ejemplo sencillo en sqlite

import sqlite3
con = sqlite3.connect(":memory:")

cur = con.cursor()
cur.execute("SELECT 1 one, 2 two")
onerow = cur.fetchone()
print('onerow', type(onerow), onerow)
cur.execute("SELECT 1 one, 2 two union all select '01' as one, '02' as two")
allrows = cur.fetchall()
print('allrows', type(allrows), allrows)

Devuelve:

onerow <class 'tuple'> (1, 2)
allrows <class 'list'> [(1, 2), ('01', '02')]

Y con el controlador InterSystems

import iris

con = iris.connect(
    hostname="localhost",
    port=1972,
    namespace="USER",
    username="_SYSTEM",
    password="SYS",
)

cur = con.cursor()
cur.execute("SELECT 1 one, 2 two")
onerow = cur.fetchone()
print("onerow", type(onerow), onerow)
cur.execute("SELECT 1 one, 2 two union all select '01' as one, '02' as two")
allrows = cur.fetchall()
print("allrows", type(allrows), allrows)

por algún motivo devuelve:

onerow <class 'iris.dbapi.DataRow'> <iris.dbapi.DataRow object at 0x104ca4e10>
allrows <class 'tuple'> ((1, 2), ('01', '02'))

¿Qué es DataRow? ¿Por qué no una tupla o al menos una lista?

Excepciones

El estandar describe una variedad de clases de excepciones que el driver debe usar en caso de que algo vaya mal. Y el driver de InterSystems no lo utiliza en absoluto, simplemente lanzando un error RunTime por cualquier razón, que no es parte del estándar de todos modos.

La aplicación debeconfiar en el tipo de excepción que ocurre, y actuar en consecuencia. Pero para el controlador InterSystems no hay ninguna diferencia. Y otra cuestión, el SQLCODE ayudaría, pero es necesario analizarlo desde mensaje de error

Conclusión

Durante la prueba encontré algunos bugs:

• Errores aleatorios ocurren en cualquier momento.<LIST ERROR> Incorrect list format, unsupported type for IRISList; Details: type detected : 32
  • Funcionará correctamente si lo intentáis de nuevo justo después del error.
• He detectado algunos errores de segmentación, no sabemos porque.
• Resultados inesperados de la función fetchone.
• Funcionamiento inesperado de la función executemany para filas con un solo resultado.
• Excepciones no implementadas en absoluto; diferentes errores deberían generar diferentes excepciones, y vuestras aplicaciones dependen de ello.
• Puede romper Embedded Python si lo instaláis junto a IRIS.
  • Debido a que Embedded Python y este controlador usan el mismo nombre, sobrescribe lo que ya está instalado con IRIS y puede romperlo.

SQLAlchemy-iris ahora soporta el driver oficial de InterSystems, pero debido a la incompatibilidad con Embedded Python y varios bugs descubiertos durante las pruebas. Instalar con este comando, con el extra definido

pip install sqlalchemy-iris[intersystems]

Y para facilitar el uso, la URL será iris+intersystems://

from sqlalchemy import Column, MetaData, Table
from sqlalchemy.sql.sqltypes import Integer, VARCHAR
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase


DATABASE_URL = "iris+intersystems://_SYSTEM:SYS@localhost:1972/USER"
engine = create_engine(DATABASE_URL, echo=True)

# Create a table metadata
metadata = MetaData()


class Base(DeclarativeBase):
    pass
def main():
    demo_table = Table(
        "demo_table",
        metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("value", VARCHAR(50)),
    )

    demo_table.drop(engine, checkfirst=True)
    demo_table.create(engine, checkfirst=True)
    with engine.connect() as conn:
        conn.execute(
            demo_table.insert(),
            [
                {"id": 1, "value": "Test"},
                {"id": 2, "value": "More"},
            ],
        )
        conn.commit()
        result = conn.execute(demo_table.select()).fetchall()
        print("result", result)


main()

Debido a errores en el controlador de InterSystems, algunas funciones pueden no funcionar como se espera. Espero que se solucionen en el futuro.