eq-sql-connector

A small python package to handle authentication and act as a generic database connector.

It supports multiple authentication flows using msal-bearer and contains a few basic database access functions including parameterized querying.

Install

Install from pypi pip install eq-sql-connector or clone and use uv sync as a developer.

Usage

For connecting to Equinor database, typically only database and url_prod is required when creating DBConnector:

from eq_sql_connector import DBConnector


connector = DBConnector(
	database="my_database",
	url_prod="my-server.database.windows.net",
)

df = connector.query("SELECT TOP 5 * FROM my_table")
print(df)

# Example with parameterized querye+
df = connector.query(
	"SELECT TOP 10 * FROM my_table WHERE asset_id IN :param",
	params=[1001, 1002, 1003],
)

connector.reset_engine()

Advanced example with an explicit Authenticator and a parameterized query:

from msal_bearer import Authenticator
from eq_sql_connector import DBConnector


authenticator = Authenticator(
	tenant_id="<your-tenant-id>",
	client_id="<your-client-id>",
	scopes=["https://database.windows.net/.default"],
	user_name="<your-user>@equinor.com",
)

connector = DBConnector(
	database="my_database",
	url_prod="my-server.database.windows.net",
	authenticator=authenticator,
)

df = connector.query(
	"SELECT TOP 10 * FROM my_table WHERE asset_id IN :param",
	params=[1001, 1002, 1003],
)

print(df.head())

# Dispose pooled connections when done
connector.reset_engine()

Testing

Run tests with coverage from the repository root:

uv run pytest --cov=src/eq_sql_connector --cov-report=term-missing

Current test suite covers:

• SQL driver detection and connection string generation
• Connector and DBConnector initialization and validation
• Query behavior (including parameter handling)
• Engine caching, reset, and pool configuration