If any example is broken, or if you’d like to add an example to this page, feel free to raise an issue on our Github repository.
Postgres Examples¶
Example of creating a Psycopg2 single connection and a connection pool
psycopg2_connect.py
:
import approzium
from approzium import AuthClient
from approzium.psycopg2 import connect
from approzium.psycopg2.pool import ThreadedConnectionPool
auth = AuthClient(
"authenticator:6001",
# This is insecure, see https://approzium.org/configuration for proper use.
disable_tls=True,
)
dsn = "host=dbmd5 dbname=db user=bob"
conn = connect(dsn, authenticator=auth)
print("Connection Established")
approzium.default_auth_client = auth
conns = ThreadedConnectionPool(1, 5, dsn)
conn = conns.getconn()
print("Connection Pool Established")
Example of creating a Asyncpg single connection and a connection pool
asyncpg_connect.py
:
import asyncio
from approzium import AuthClient
from approzium.asyncpg import connect
from approzium.asyncpg.pool import create_pool
auth = AuthClient(
"authenticator:6001",
# This is insecure, see https://approzium.org/configuration for proper use.
disable_tls=True,
)
async def run():
conn = await connect(user="bob", database="db", host="dbmd5", authenticator=auth)
print("Connection Established!")
await conn.fetch("""SELECT 1""")
await conn.close()
pool = await create_pool(
user="bob", database="db", host="dbmd5", 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 Examples¶
Example of creating a MySQL Connector single connection and a connection pool
mysql_connector_connect.py
:
from approzium import AuthClient
from approzium.mysql.connector import connect
from approzium.mysql.connector.pooling import MySQLConnectionPool
auth = AuthClient(
"authenticator:6001",
# This is insecure, see https://approzium.org/configuration for proper use.
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="dbmysqlsha1",
authenticator=auth,
use_pure=True,
)
print("Connection Pool Established")
conn = cnxpool.get_connection()
cur = conn.cursor()
cur.execute("SELECT 1")
print(result)
Example of creating a PyMySQL single connection:
pymysql_connect.py
:
from approzium import AuthClient
from approzium.pymysql import connect
auth = AuthClient(
"authenticator:6001",
# This is insecure, see https://approzium.org/configuration for proper use.
disable_tls=True,
)
conn = connect(host="dbmysqlsha1", user="bob", db="db", authenticator=auth)
with conn.cursor() as cursor:
cursor.execute("SELECT 1")
result = cursor.fetchone()
print(result)
conn.close()
Opentelemetry Integration Examples¶
psycopg2_opentelemetry.py
:
"""This is an example that shows Approzium Opentelemetry integration. It also
integrates with a Jaeger service to export and view generated traces.
"""
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import approzium
import approzium.opentelemetry
from approzium.psycopg2 import connect
auth = approzium.AuthClient("authenticator:6001")
approzium.default_auth_client = auth
trace.set_tracer_provider(
TracerProvider(resource=Resource.create({SERVICE_NAME: "approzium_service"}))
)
tracer = trace.get_tracer(__name__)
jaeger_exporter = JaegerExporter(agent_host_name="jaeger", agent_port=6831)
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(jaeger_exporter))
approzium.opentelemetry.instrument()
Psycopg2Instrumentor().instrument()
cnx = connect("host=dbmd5 dbname=db user=bob")
cursor = cnx.cursor()
with tracer.start_as_current_span("foo"):
with tracer.start_as_current_span("bar"):
print("Hello world!")
cursor.execute("SELECT 1")
cursor.close()
cnx.close()
If you are not using Opentelemetry, you can obtain the same attribution info manually:
psycopg2_attribution_info.py
:
"""This example shows usage of the AuthClient.attribution_info feature. It uses
a Psycopg2 connection but the same functionality is available in any supported
database driver.
"""
import approzium
from approzium.psycopg2 import connect
auth = approzium.AuthClient("authenticator:6001")
print(auth.attribution_info)
# {'authenticator_address': 'authenticator:6001',
# 'iam_arn': 'arn:aws:iam::*******:user/****',
# 'authenticated': False,
# 'num_connections': 0
# ... additional info if available (ex: EC2 instance metadata)
# }
approzium.default_auth_client = auth
dsn = "host=dbmd5 dbname=db user=bob"
conn = connect(dsn)
print(auth.attribution_info)
# {'authenticator_address': 'authenticator:6001',
# 'iam_arn': 'arn:aws:iam::*******:user/****',
# 'authenticated': True,
# 'num_connections': 1
# ...
# }