Examples

Python Examples

Postgres Drivers

Psycopg2

sdk/python/examples/psycopg2_connect.py

import approzium
from approzium.psycopg2 import connect
from approzium.psycopg2.pool import ThreadedConnectionPool

auth = approzium.AuthClient("authenticator:6001", disable_tls=True)
approzium.default_auth_client = auth
dsn = "host=dbmd5 dbname=db user=bob"
conn = connect(dsn)
print("Connection Established")

conns = ThreadedConnectionPool(1, 5, dsn)
conn = conns.getconn()
print("Connection Pool Established")

Asyncpg

sdk/python/examples/asyncpg_connect.py

import asyncio

from approzium import Authenticator
from approzium.asyncpg import connect
from approzium.asyncpg.pool import create_pool

auth = Authenticator("authenticator:6001", disable_tls=True)


async def run():
    conn = await connect(user="bob", database="db", host="host", authenticator=auth)
    print("Connection Established!")
    await conn.fetch("""SELECT 1""")
    await conn.close()

    pool = await create_pool(user="bob", database="db", host="host", authenticator=auth)
    print("Connection Established!")
    async with pool.acquire() as conn:
        await conn.fetch("""SELECT 1""")


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

MySQL Drivers

MySQL Connector

sdk/python/examples/mysql_connector_connect.py

import approzium
from approzium.mysql.connector import connect
from approzium.mysql.connector.pooling import MySQLConnectionPool

auth = approzium.AuthClient("authenticator:6001", disable_tls=True)
conn = connect(user="bob", authenticator=auth, host="dbmysql", use_pure=True)
print("Connection Established")

cur = conn.cursor()
cur.execute("SELECT 1")
result = next(cur)
print(result)

cnxpool = MySQLConnectionPool(
    pool_name="mypool", pool_size=3, user="bob", host="dbmysql", authenticator=auth
)
print("Connection Pool Established")
conn = cnxpool.get_connection()
cur = conn.cursor()
cur.execute("SELECT 1")
print(result)

Using TLS

sdk/python/examples/mysql_connector_connect.py

import approzium
from approzium.mysql.connector import connect
from approzium.mysql.connector.pooling import MySQLConnectionPool

auth = approzium.AuthClient(
    "authenticator:6001",
    tls_config=approzium.TLSConfig(
        trusted_certs=environ.get("TEST_CERT_DIR") + "/approzium.pem",
        client_cert=environ.get("TEST_CERT_DIR") + "/client.pem",
        client_key=environ.get("TEST_CERT_DIR") + "/client.key",
    ),
)