A SQLAlchemy dialect for connecting to a Theseus Flight SQL server with ADBC
Project description
SQLAlchemy Theseus Dialect - powered by Arrow Flight SQL ADBC
Basic SQLAlchemy dialect for the Theseus Data Processing Engine
Installation
Option 1 - from PyPi
$ pip install sqlalchemy-theseus-dialect
Option 2 - from source - for development
git clone https://github.com/prmoore77/sqlalchemy-theseus-dialect
cd sqlalchemy-theseus-dialect
# Create the virtual environment
python3 -m venv .venv
# Activate the virtual environment
. .venv/bin/activate
# Upgrade pip, setuptools, and wheel
pip install --upgrade pip setuptools wheel
# Install SQLAlchemy Theseus Dialect - in editable mode with dev dependencies
pip install --editable .[dev]
Note
For the following commands - if you running from source and using --editable
mode (for development purposes) - you will need to set the PYTHONPATH environment variable as follows:
export PYTHONPATH=$(pwd)/src
Usage
Once you've installed this package, you should be able to just use it, as SQLAlchemy does a python path search
Ensure you have access to a Theseus engine
Connect with the SQLAlchemy Theseus Dialect
import logging
from sqlalchemy import create_engine, MetaData, Table, select, Column, text, Integer, String, Sequence
from sqlalchemy.orm import Session
from sqlalchemy.engine.url import URL
# Setup logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
def main():
# Build the URL
url = URL.create(drivername="theseus",
host="theseus-gateway.vice.svc.cluster.local",
port=11234,
query={"disableCertificateVerification": "True",
"useEncryption": "False"
}
)
print(f"Database URL: {url}")
engine = create_engine(url=url)
metadata = MetaData()
metadata.reflect(bind=engine)
for table_name in metadata.tables:
print(f"Table name: {table_name}")
with Session(bind=engine) as session:
# Execute some raw SQL (this assumes you have a registered table called: "fake")
results = session.execute(statement=text("SELECT * FROM fake")).fetchall()
print(results)
# Try a SQLAlchemy table select
fake: Table = metadata.tables["fake"]
stmt = select(fake.c.name)
results = session.execute(statement=stmt).fetchall()
print(results)
if __name__ == "__main__":
main()
Credits
Much code and inspiration was taken from repo: https://github.com/Mause/duckdb_engine
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for sqlalchemy_theseus_dialect-0.0.2.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | aa2ffa7978792e9995be264e9e758b467132aea36a522ff4752766e76e47d622 |
|
MD5 | fc1cdec034b57781894265cefb80596e |
|
BLAKE2b-256 | 64da4aa7923db260684088673e089bbf7656d9dd1c6580566c22a511a1b32c5d |
Close
Hashes for sqlalchemy_theseus_dialect-0.0.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a4c707ea93a7fc95f6d0d0d3568c875fd083eb6b0e1251e3b9e2e29b8afc2b0d |
|
MD5 | 13665da8147decf46688c405120944d8 |
|
BLAKE2b-256 | 08324c65b822346076976561518958119e4a2076b08eee93865eed2a931942aa |