tessl/pypi-connectorx

Load data from databases to dataframes, the fastest way.

1.04x

Quality

Pending

Does it follow best practices?

Impact

86%

1.04x

Average score across 10 eval scenarios

Overview

Eval results

Files

Connection Management

Name: tessl/pypi-connectorx
Rating: 0.86 (1 reviews)
Author: tessl

Helper utilities for building and managing database connection strings across different database backends. ConnectorX provides the ConnectionUrl class to simplify connection string construction and ensure proper formatting for various database systems.

Capabilities

ConnectionUrl Class

Generic connection URL builder that supports multiple database backends with type safety.

class ConnectionUrl(Generic[_BackendT], str):
    """
    Helper class to build database connection string URLs.
    Inherits from str, so can be used anywhere a connection string is expected.
    """
    
    # SQLite constructor
    def __new__(
        cls,
        *,
        backend: Literal["sqlite"],
        db_path: str | Path,
    ) -> ConnectionUrl[Literal["sqlite"]]:
        """Build SQLite connection URL."""
    
    # BigQuery constructor  
    def __new__(
        cls,
        *,
        backend: Literal["bigquery"],
        db_path: str | Path,
    ) -> ConnectionUrl[Literal["bigquery"]]:
        """Build BigQuery connection URL."""
    
    # Server-based database constructor
    def __new__(
        cls,
        *,
        backend: _ServerBackendT,
        username: str,
        password: str = "",
        server: str,
        port: int,
        database: str = "",
        database_options: dict[str, str] | None = None,
    ) -> ConnectionUrl[_ServerBackendT]:
        """Build server-based database connection URL."""
    
    # Raw connection string constructor
    def __new__(
        cls,
        raw_connection: str,
    ) -> ConnectionUrl:
        """Build connection from raw connection string."""

Supported Backend Types

_ServerBackendT = TypeVar(
    "_ServerBackendT",
    bound=Literal[
        "redshift",
        "clickhouse", 
        "postgres",
        "postgresql",
        "mysql",
        "mssql",
        "oracle",
        "duckdb",
    ],
)

Usage Examples

SQLite Connections

from connectorx import ConnectionUrl
from pathlib import Path

# Using string path
sqlite_url = ConnectionUrl(
    backend="sqlite",
    db_path="/path/to/database.db"
)

# Using Path object
db_path = Path("./data/sample.sqlite")
sqlite_url = ConnectionUrl(
    backend="sqlite", 
    db_path=db_path
)

# Result: "sqlite:///path/to/database.db"

BigQuery Connections

# BigQuery with service account key file
bq_url = ConnectionUrl(
    backend="bigquery",
    db_path="/path/to/service-account-key.json"
)

# Result: "bigquery:///path/to/service-account-key.json"

PostgreSQL Connections

# Basic PostgreSQL connection
postgres_url = ConnectionUrl(
    backend="postgresql",
    username="myuser",
    password="mypassword", 
    server="localhost",
    port=5432,
    database="mydb"
)

# With connection options
postgres_url = ConnectionUrl(
    backend="postgresql",
    username="myuser",
    password="mypassword",
    server="postgres.example.com",
    port=5432,
    database="production_db",
    database_options={
        "sslmode": "require",
        "application_name": "connectorx_app"
    }
)

# Result: "postgresql://myuser:mypassword@postgres.example.com:5432/production_db?sslmode=require&application_name=connectorx_app"

MySQL Connections

mysql_url = ConnectionUrl(
    backend="mysql",
    username="root",
    password="secret",
    server="mysql.example.com", 
    port=3306,
    database="analytics"
)

# Result: "mysql://root:secret@mysql.example.com:3306/analytics"

SQL Server Connections

mssql_url = ConnectionUrl(
    backend="mssql",
    username="sa",
    password="Password123!",
    server="sqlserver.example.com",
    port=1433,
    database="AdventureWorks",
    database_options={
        "TrustServerCertificate": "true",
        "Encrypt": "false"
    }
)

Oracle Connections

oracle_url = ConnectionUrl(
    backend="oracle",
    username="hr",
    password="oracle",
    server="oracle.example.com",
    port=1521,
    database="ORCL"
)

Raw Connection Strings

# Use existing connection string
raw_conn = "postgresql://user:pass@host:port/db?sslmode=require"
conn_url = ConnectionUrl(raw_conn)

# ConnectionUrl inherits from str, so it works everywhere
import connectorx as cx
df = cx.read_sql(conn_url, "SELECT * FROM table")

Advanced Features

Connection String Rewriting

ConnectorX automatically rewrites certain connection strings for compatibility:

# Redshift connections are rewritten to use PostgreSQL driver
redshift_url = ConnectionUrl(
    backend="redshift",
    username="user",
    password="pass", 
    server="cluster.redshift.amazonaws.com",
    port=5439,
    database="dev"
)
# Internally rewritten to: postgresql://user:pass@cluster.redshift.amazonaws.com:5439/dev
# Protocol automatically set to "cursor"

# ClickHouse connections are rewritten to use MySQL driver  
clickhouse_url = ConnectionUrl(
    backend="clickhouse",
    username="default",
    password="",
    server="clickhouse.example.com",
    port=9000,
    database="default"
)
# Internally rewritten to: mysql://default:@clickhouse.example.com:9000/default
# Protocol automatically set to "text"

Type Safety

ConnectionUrl provides type safety for different backends:

from typing import Literal

def connect_to_postgres(url: ConnectionUrl[Literal["postgresql"]]) -> None:
    # Function specifically expects PostgreSQL connection
    pass

postgres_url = ConnectionUrl(backend="postgresql", ...)  # Type: ConnectionUrl[Literal["postgresql"]]
mysql_url = ConnectionUrl(backend="mysql", ...)         # Type: ConnectionUrl[Literal["mysql"]] 

connect_to_postgres(postgres_url)  # ✓ Type check passes
connect_to_postgres(mysql_url)     # ✗ Type error

Integration with ConnectorX Functions

ConnectionUrl objects work seamlessly with all ConnectorX functions:

from connectorx import ConnectionUrl, read_sql, get_meta, partition_sql

# Build connection
conn = ConnectionUrl(
    backend="postgresql",
    username="user",
    password="pass",
    server="localhost", 
    port=5432,
    database="mydb"
)

# Use with all ConnectorX functions
df = read_sql(conn, "SELECT * FROM table")
meta = get_meta(conn, "SELECT * FROM table")
queries = partition_sql(conn, "SELECT * FROM table", "id", 4)

Connection String Formats

Supported URL Schemes

postgresql:// - PostgreSQL databases
mysql:// - MySQL and MariaDB databases
sqlite:// - SQLite database files
mssql:// - Microsoft SQL Server
oracle:// - Oracle databases
bigquery:// - Google BigQuery
redshift:// - Amazon Redshift (rewritten to PostgreSQL)
clickhouse:// - ClickHouse (rewritten to MySQL)

URL Encoding

ConnectionUrl automatically handles URL encoding for special characters:

# Password with special characters
conn = ConnectionUrl(
    backend="postgresql",
    username="user",
    password="p@ssw0rd!#$",  # Special characters automatically encoded
    server="localhost",
    port=5432,
    database="mydb"
)

Common Connection Patterns

# Local development
local_pg = ConnectionUrl(
    backend="postgresql",
    username="postgres", 
    password="postgres",
    server="localhost",
    port=5432,
    database="dev_db"
)

# Production with SSL
prod_pg = ConnectionUrl(
    backend="postgresql", 
    username="app_user",
    password="secure_password",
    server="prod-db.company.com",
    port=5432,
    database="production",
    database_options={"sslmode": "require"}
)

# Read-only replica
replica_pg = ConnectionUrl(
    backend="postgresql",
    username="readonly_user", 
    password="readonly_pass",
    server="replica.company.com",
    port=5432,
    database="production",
    database_options={
        "sslmode": "require",
        "default_transaction_read_only": "on"
    }
)

Install with Tessl CLI

npx tessl i tessl/pypi-connectorx

tessl/pypi-connectorx