No SSL

import drda

conn = drda.connect(host='serverhost', database='dbname', user='user', password='password', port=xxxxx, timeout=30)
cur = conn.cursor()
cur.execute('select * from foo where name=?', ['alice'])
for r in cur.fetchall():
    print(r[0], r[1])
conn.close()

With SSL connection

import drda

conn = drda.connect(host='serverhost', database='dbname', use_ssl=True, user='user', password='password', port=xxxxx)
cur = conn.cursor()
cur.execute('select * from foo where name=?', ['alice'])
for r in cur.fetchall():
    print(r[0], r[1])
conn.close()

With SSL and CA certificate

To connect to a server with a self-signed certificate or custom CA (such as IBM Db2 on Cloud), pass the server’s CA certificate file path via ssl_client_cert_path:

import drda

conn = drda.connect(host='serverhost', database='dbname', use_ssl=True, ssl_client_cert_path='/path/to/ca-cert.crt', user='user', password='password', port=xxxxx)
cur = conn.cursor()
cur.execute('select * from foo where name=?', ['alice'])
for r in cur.fetchall():
    print(r[0], r[1])
conn.close()

Context Manager and Cursor Iteration

Connections and cursors support Python context managers (with statement), and cursors can be iterated directly without calling fetchall():

import drda

with drda.connect(host='serverhost', database='dbname', user='user', password='password', port=xxxxx) as conn:
    with conn.cursor() as cur:
        cur.execute('select * from foo where name=?', ['alice'])
        for r in cur:
            print(r[0], r[1])

Transactions

Transactions can be explicitly committed or rolled back:

cur.execute("insert into foo values (?, ?)", [1, 'alice'])
conn.commit()

# Or rollback changes:
# conn.rollback()

AsyncIO

import asyncio
import drda.aio

async def main():
    async with await drda.aio.connect(host='serverhost', database='dbname', user='user', password='password', port=xxxxx, timeout=30) as conn:
        async with conn.cursor() as cur:
            await cur.execute('select * from foo where name=?', ['alice'])
            async for r in cur:
                print(r[0], r[1])

asyncio.run(main())

Dictionary Cursor

To fetch query results as Python dictionaries instead of tuples:

import drda

with drda.connect(host='serverhost', database='dbname', user='user', password='password', port=xxxxx) as conn:
    with conn.cursor(drda.DictCursor) as cur:
        cur.execute('select * from foo where name=?', ['alice'])
        for r in cur:
            print(r['ID'], r['NAME'])

For asyncio, use drda.aio.AsyncDictCursor:

import asyncio
import drda.aio

async def main():
    async with await drda.aio.connect(host='serverhost', database='dbname', user='user', password='password', port=xxxxx) as conn:
        async with conn.cursor(drda.aio.AsyncDictCursor) as cur:
            await cur.execute('select * from foo where name=?', ['alice'])
            async for r in cur:
                print(r['ID'], r['NAME'])

asyncio.run(main())