DataEngine

A lightweight Python wrapper around SQLAlchemy, pandas, pyodbc, psycopg2, and pymongo that provides a unified interface for connecting to SQL Server, PostgreSQL, and MongoDB.

Connection strings are stored in a single database.env file and loaded at runtime — no credentials in code.

Installation

Requires Python >= 3.10 and an active virtual environment.

pip install Python-DataEngine

To force a reinstall:

pip install --force-reinstall Python-DataEngine

SQL Server prerequisite

DataEngine uses pyodbc and requires a Microsoft ODBC driver. Supported drivers in order of preference:

1. ODBC Driver 18 for SQL Server (recommended)
2. ODBC Driver 17 for SQL Server
3. SQL Server Native Client 11.0
4. SQL Server

Download: https://go.microsoft.com/fwlink/?linkid=2266640

Quick Start

import DataEngine

DataEngine.initialize()  # loads database.env and creates connection objects

db = DataEngine.alchemyObjects["my_connection"]
rows = db.query("SELECT TOP 10 * FROM dbo.MyTable")

Connection Configuration

Connections are stored in a file named database.env in your working directory. Each entry is a JSON object under the databases key.

databases = '{"my_connection": {"type": "mssql", "server": "MYSERVER", "database": "MyDB", "UN": "", "PW": "", "trusted": "yes"}}'

All values must be strings. Use the interactive builder to create this file:

DataEngine.connectionStringBuilder()

Connection document schema

SQL Server — SqlConnectionObject

Supports three authentication modes, selected automatically based on the values in database.env:

Azure AD Interactive caches its token for approximately one hour, after which it silently re-authenticates using the cached session.

Methods

Example

import DataEngine

DataEngine.initialize()
sql = DataEngine.alchemyObjects["sql"]

# Query to a list of rows
rows = sql.query("SELECT id, name FROM dbo.Users WHERE active = 1")

# Query to a DataFrame
df = sql.getTable("SELECT * FROM dbo.Sales")

# Stream a large result set in chunks
for chunk in sql.chunkTable("SELECT * FROM dbo.BigTable", chunksize=10000):
    process(chunk)

# Execute a stored procedure
sql.executeProcedure("dbo.usp_RefreshSummary")

# Insert and get the new row ID
new_id = sql.interop("INSERT INTO dbo.Log (message) VALUES ('started')")

PostgreSQL — PgConnectionObject

Connects via psycopg2. Always requires UN and PW.

Methods

The same interface as SqlConnectionObject: query, getTable, chunkTable, queryStream, execute, executeProcedure, truncateTable.

Example

import DataEngine

DataEngine.initialize()
pg = DataEngine.alchemyObjects["postgres"]

df = pg.getTable("SELECT * FROM public.orders WHERE status = 'open'")

MongoDB — MongoConnectionObject

Connects via pymongo using a standard mongodb:// URI with authSource=admin.

Methods

Example

import DataEngine

DataEngine.initialize()
mongo = DataEngine.alchemyObjects["mongo"]

cursor = mongo.query("users", {"active": True})
for doc in cursor:
    print(doc)

result = mongo.aggregate("sales", [
    {"$match": {"year": 2025}},
    {"$group": {"_id": "$region", "total": {"$sum": "$amount"}}}
])

Module-level API

After initialize(), connections are available in:

DataEngine.alchemyObjects   # dict of connection objects, keyed by name
DataEngine.alchemyConnections  # dict of raw connection string dicts

Development

Install dev dependencies:

pip install -r requirements-dev.txt

Run the test suite (no live database required — all tests are mocked):

pytest