SQLAchemy-iris with the latest version Python driver

04512

After so many years of waiting, we finally got an official driver available on Pypi

Additionally, found JDBC driver finally available on Maven already for 3 months,  and .Net driver on Nuget more than a month.

 As an author of so many implementations of IRIS support for various Python libraries, I wanted to check it. Implementation of DB-API means that it should be replaceable and at least functions defined in the standard. The only difference should be in SQL.

And the beauty of using already existing libraries, that they already implemented other databases by using DB-API standard, and these libraries already expect how driver should work.

I decided to test InterSystems official driver by implementing its support in SQLAlchemy-iris library.

executemany

Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings found in the sequence seq_of_parameters.

Very helpful function, whish let insert multiple rows at once. Let’s start with a simple example

<span>import</span> iris
host = <span>"localhost"</span>
port = <span>1972</span>
namespace = <span>"USER"</span>
username = <span>"_SYSTEM"</span>
password = <span>"SYS"</span>
conn = iris.connect(
    host,
    port,
    namespace,
    username,
    password,
)
<span>with</span> conn.cursor() <span>as</span> cursor:
    cursor = conn.cursor()
    res = cursor.execute(<span>"DROP TABLE IF EXISTS test"</span>)
    res = cursor.execute(
        <span>""" CREATE TABLE test ( id IDENTITY NOT NULL, value VARCHAR(50) ) WITH %CLASSPARAMETER ALLOWIDENTITYINSERT = 1 """</span>
    )
    cursor = conn.cursor()
    res = cursor.executemany(
        <span>"INSERT INTO test (id, value) VALUES (?, ?)"</span>, [
            (<span>1</span>, <span>'val1'</span>),
            (<span>2</span>, <span>'val2'</span>),
            (<span>3</span>, <span>'val3'</span>),
            (<span>4</span>, <span>'val4'</span>),
        ]
    )
<span>import</span> iris

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

<span>with</span> conn.cursor() <span>as</span> cursor:
    cursor = conn.cursor()

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

    cursor = conn.cursor()
    res = cursor.executemany(
        <span>"INSERT INTO test (id, value) VALUES (?, ?)"</span>, [
            (<span>1</span>, <span>'val1'</span>),
            (<span>2</span>, <span>'val2'</span>),
            (<span>3</span>, <span>'val3'</span>),
            (<span>4</span>, <span>'val4'</span>),
        ]
    )
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'),
        ]
    )

This is working fine, but what if we need to insert only one value per row.

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

This unfortunately leads to an unexpected exception

RuntimeError: Cannot use list/tuple for single values

By some reason, one value per row is allowed, and InterSystems requires using a different way

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

This way it’s working fine

fetchone

Fetch the next row of a query result set, returning a single sequence, or None when no more data is available.

For instance simple example on sqlite

<span>import</span> sqlite3
con = sqlite3.connect(<span>":memory:"</span>)
cur = con.cursor()
cur.execute(<span>"SELECT 1 one, 2 two"</span>)
onerow = cur.fetchone()
print(<span>'onerow'</span>, type(onerow), onerow)
cur.execute(<span>"SELECT 1 one, 2 two union all select '01' as one, '02' as two"</span>)
allrows = cur.fetchall()
print(<span>'allrows'</span>, type(allrows), allrows)
<span>import</span> sqlite3
con = sqlite3.connect(<span>":memory:"</span>)

cur = con.cursor()
cur.execute(<span>"SELECT 1 one, 2 two"</span>)
onerow = cur.fetchone()
print(<span>'onerow'</span>, type(onerow), onerow)
cur.execute(<span>"SELECT 1 one, 2 two union all select '01' as one, '02' as two"</span>)
allrows = cur.fetchall()
print(<span>'allrows'</span>, type(allrows), allrows)
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)

gives

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

And with InterSystems driver

<span>import</span> iris
con = iris.connect(
    hostname=<span>"localhost"</span>,
    port=<span>1972</span>,
    namespace=<span>"USER"</span>,
    username=<span>"_SYSTEM"</span>,
    password=<span>"SYS"</span>,
)
cur = con.cursor()
cur.execute(<span>"SELECT 1 one, 2 two"</span>)
onerow = cur.fetchone()
print(<span>"onerow"</span>, type(onerow), onerow)
cur.execute(<span>"SELECT 1 one, 2 two union all select '01' as one, '02' as two"</span>)
allrows = cur.fetchall()
print(<span>"allrows"</span>, type(allrows), allrows)
<span>import</span> iris

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

cur = con.cursor()
cur.execute(<span>"SELECT 1 one, 2 two"</span>)
onerow = cur.fetchone()
print(<span>"onerow"</span>, type(onerow), onerow)
cur.execute(<span>"SELECT 1 one, 2 two union all select '01' as one, '02' as two"</span>)
allrows = cur.fetchall()
print(<span>"allrows"</span>, type(allrows), allrows)
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)

by some reasons gives

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

What is DataRow, why not tuple or at least a list

Exceptions

Standard describes a variety of exception classes that the driver is supposed to use, in case if something is wrong. And the InterSystems driver does not use it at all, just raising RunTime error for any reason, which is not part of the standard anyway.

Application may rely on the exception type happening, and behave accordingly. But InterSystems driver does not provide any difference. And another issue, SQLCODE would help, but it needs to be parsed out of error message

Conclusion

So, during testing I found multiple bugs

  • Random errors happening at any time <LIST ERROR> Incorrect list format, unsupported type for IRISList; Details: type detected : 32
    • will work ok, if you try again right after the error
  • Caught some segmentation faults, don’t even know how it happens
  • Unexpected result from fetchone function
  • Unexpected way of working of executemany function, for one value rows
  • Exceptions not implemented at all, different errors should raise different exceptions, and applications rely on it
  • Can break Embedded Python if installed next to IRIS
    • due to the same name used by Embedded Python and this driver, it overrides what’s already installed with IRIS and may break it

 

SQLAlchemy-iris now supports the official InterSystems driver, but due to incompatibility with Embedded Python and several bugs discovered during testing. Install with this command, with the defined extra

pip install sqlalchemy-iris[intersystems]

And simple usage, URL should be iris+intersystems://

<span>from</span> sqlalchemy <span>import</span> Column, MetaData, Table
<span>from</span> sqlalchemy.sql.sqltypes <span>import</span> Integer, VARCHAR
<span>from</span> sqlalchemy <span>import</span> create_engine
<span>from</span> sqlalchemy.orm <span>import</span> DeclarativeBase
DATABASE_URL = <span>"iris+intersystems://_SYSTEM:SYS@localhost:1972/USER"</span>
engine = create_engine(DATABASE_URL, echo=<span>True</span>)
<span># Create a table metadata</span>
metadata = MetaData()
<span><span>class</span> <span>Base</span><span>(DeclarativeBase)</span>:</span>
    <span>pass</span>
<span><span>def</span> <span>main</span><span>()</span>:</span>
    demo_table = Table(
        <span>"demo_table"</span>,
        metadata,
        Column(<span>"id"</span>, Integer, primary_key=<span>True</span>, autoincrement=<span>True</span>),
        Column(<span>"value"</span>, VARCHAR(<span>50</span>)),
    )
    demo_table.drop(engine, checkfirst=<span>True</span>)
    demo_table.create(engine, checkfirst=<span>True</span>)
    <span>with</span> engine.connect() <span>as</span> conn:
        conn.execute(
            demo_table.insert(),
            [
                {<span>"id"</span>: <span>1</span>, <span>"value"</span>: <span>"Test"</span>},
                {<span>"id"</span>: <span>2</span>, <span>"value"</span>: <span>"More"</span>},
            ],
        )
        conn.commit()
        result = conn.execute(demo_table.select()).fetchall()
        print(<span>"result"</span>, result)
main()
<span>from</span> sqlalchemy <span>import</span> Column, MetaData, Table
<span>from</span> sqlalchemy.sql.sqltypes <span>import</span> Integer, VARCHAR
<span>from</span> sqlalchemy <span>import</span> create_engine
<span>from</span> sqlalchemy.orm <span>import</span> DeclarativeBase


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

<span># Create a table metadata</span>
metadata = MetaData()


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

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


main()
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()

Due to bugs in InterSystems driver, some features may not work as expected. And I hope it will be fixed in the future

原文链接：SQLAchemy-iris with the latest version Python driver

展开阅读全文
© 版权声明
文章版权声明 1、本网站名称：拾光赋
2、本站永久网址：https://www.blogs.ink
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长QQ：805375623进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END
Python（EN）
# python# programming# opensource# sql
喜欢就支持一下吧
点赞12 分享
When you procrastinate, you become a slave to yesterday.
拖延会让你成为昨天的奴隶
评论 抢沙发

请登录后发表评论

    暂无评论内容

文章目录
kity的头像-拾光赋
文章目录
今日剩余 12.8%
本周剩余 30.4%
本月剩余 40.4%
本年剩余 70.4%
最近评论
杀人蜂的头像-拾光赋

杀人蜂5天前0

我很好奇这个网站能提供什么
Vito的头像-拾光赋

Vito6个月前2

你就是我心中的那首忐忑，总是让我惊心动魄
Jason.Meng的头像-拾光赋钻石会员

Jason.Meng徽章-人气大使-拾光赋6个月前2

总是会在人生的不同阶段反复爱上大佬的文章~
每日一言
pride relates more to our opinion of ourselves, vanity to what we would have others think of us.
骄傲多半涉及我们自己怎样看待自己，而虚荣则涉及我们想别人怎样看我们