sqlalchemy-jdbcapi

import os from sqlalchemy import create_engine, text # Ensure JAVA_HOME is set for JPype to find a JVM # Example: os.environ['JAVA_HOME'] = '/path/to/your/jdk' # Example for PostgreSQL via JDBC # Replace with your actual database details or environment variables DB_USER = os.environ.get('JDBC_DB_USER', 'your_user') DB_PASS = os.environ.get('JDBC_DB_PASS', 'your_password') DB_HOST = os.environ.get('JDBC_DB_HOST', 'localhost') DB_PORT = os.environ.get('JDBC_DB_PORT', '5432') DB_NAME = os.environ.get('JDBC_DB_NAME', 'your_database') # The dialect string uses 'jdbcapi+' followed by the driver name # For PostgreSQL, it's 'postgresql'. For MySQL, 'mysql', etc. jdbc_url = f"jdbcapi+postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" try: engine = create_engine(jdbc_url) with engine.connect() as connection: result = connection.execute(text("SELECT 1")) print(f"Connection successful, result: {result.scalar()}") # Example with asyncio (requires sqlalchemy-jdbcapi>=2.2.1 and asyncpg, aiomysql, etc.) # from sqlalchemy.ext.asyncio import create_async_engine # async_jdbc_url = f"jdbcapi+postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" # async_engine = create_async_engine(async_jdbc_url) # async def run_async_query(): # async with async_engine.connect() as conn: # result = await conn.execute(text("SELECT 2")) # print(f"Async connection successful, result: {result.scalar()}") # import asyncio # asyncio.run(run_async_query()) except Exception as e: print(f"Error connecting to database: {e}") finally: # JPype requires explicit JVM shutdown in some contexts or when done try: from jpype import isJVMStarted, shutdownJVM if isJVMStarted(): shutdownJVM() print("JVM shut down.") except ImportError: pass # JPype might not be installed or available