e6data Python Connector

Package version and extras are defined in setup.py.

Introduction

The e6data Connector for Python provides an interface for writing Python applications that can connect to e6data and perform operations. It includes routing support for blue-green deployments. Active query handles remain tied to the engine that accepted them; failover does not guarantee uninterrupted results.

Optional native asyncio API

This branch adds native async connection, cursor, pooling and SQLAlchemy support for Python 3.11+. Release qualification remains incomplete. Install this branch's built package with [async] or [async-sqlalchemy]; these instructions do not assert that the feature is published.

from e6data_python_connector.aio import AsyncConnection

async def query(config, sql):
    async with AsyncConnection(**config) as connection:
        async with connection.cursor() as cursor:
            await cursor.execute(sql)
            return await cursor.fetchall()

config is application-supplied connection configuration. OAuth requires verified TLS. Native fetchone() preserves the one-row outer list, while SQLAlchemy e6data+asyncio adapts it to a normal row. Ambiguous submission and incomplete result errors must not be handled by blindly retrying the query. Use context managers for bounded cleanup.

See Async API for the complete API map, deadlines, pooling, SQLAlchemy and test configuration, OAuth lifecycle for renewal and failure behavior, and the application example.

The new CI keeps the full-package coverage denominator and a greater-than-80% gate. The initial baseline was about 19%, so focused async test success is not a release pass. Real-service tests require explicit configuration and are qualified separately.

Dependencies

Make sure to install below dependencies and wheel before install e6data-python-connector.

# Amazon Linux / CentOS dependencies
yum install python3-devel gcc-c++ -y

# Ubuntu/Debian dependencies
apt install python3-dev g++ -y

# Windows dependencies
# Install Visual C++ Build Tools from:
# https://visualstudio.microsoft.com/visual-cpp-build-tools/
# Select the "Desktop Development with C++" option during installation.

# Pip dependencies
pip install wheel

To install the Python package, use the command below:

pip install --no-cache-dir e6data-python-connector

Prerequisites

• Open Inbound Port 80 in the Engine Cluster.
• Limit access to Port 80 according to your organizational security policy. Public access is not encouraged.
• Access Token generated in the e6data console.

Create a Connection

Use your e6data Email ID as the username and your access token as the password.

from e6data_python_connector import Connection
# For connection pooling (recommended for concurrent operations)
from e6data_python_connector import ConnectionPool

username = '<username>'  # Your e6data Email ID.
password = '<password>'  # Access Token generated in the e6data console.

host = '<host>'  # IP address or hostname of the cluster to be used.
database = '<database>'  # Database to perform the query on.
port = 80  # Port of the e6data engine.
catalog_name = '<catalog_name>'

# Single connection (for simple, single-threaded use)
conn = Connection(
    host=host,
    port=port,
    username=username,
    database=database,
    password=password
)

# Or use connection pool (for concurrent/multi-threaded use)
pool = ConnectionPool(
    min_size=2,
    max_size=10,
    host=host,
    port=port,
    username=username,
    database=database,
    password=password
)

Connection Parameters

The Connection class supports the following parameters:

Authenticating with OAuth 2.0

As an alternative to username and password, a connection can authenticate with an OAuth 2.0 access token. The connector obtains a token using the client-credentials grant, caches it, and refreshes it shortly before it expires.

Matching client-credentials configurations share one token cache and concurrent refresh across sync and async connections, separate pools and threads in the same process. Connection and transport ownership remain local. See the OAuth lifecycle for matching settings and expiry behavior.

conn = Connection(
    host=host,
    port=443,
    database=database,
    cluster_name='<cluster-name>',
    secure=True,
    client_id='<client_id>',
    client_secret='<client_secret>',
    token_url='https://<your-workspace>/oauth2/token',
)

If you mint tokens yourself, pass one directly instead. The connector will not refresh it, so a long-lived connection may outlive the token:

conn = Connection(
    host=host, port=443, database=database, cluster_name='<cluster-name>',
    secure=True, access_token='<token>',
)

Supply exactly one authentication method. Passing both username/password and OAuth settings raises a ValueError rather than picking one, so a stale value left in a config file cannot quietly win.

OAuth sends authorization: Bearer <token> on every RPC. It does not create an authentication session; protobuf username, password and session ID fields remain empty. The server stack must support and enable this bearer-only flow.

Use secure=True with your TLS endpoint and an HTTPS token_url to protect tokens and client credentials. OAuth does not enable TLS automatically; secure defaults to False. Set cluster_name to identify the target cluster explicitly.

With auto_resume=True (the default), OAuth can recover an initial query prepare that returns the exact suspended-cluster response. This requires compatible, enabled server components that authorize resume and report readiness. The connector performs one recovery sequence and retries prepare after readiness. It never replays query execution, metadata retrieval or result fetching, and it does not treat generic connection failures as suspension.

Recovery has a shared 300-second default deadline covering lock waits, token acquisition, status/resume calls, polling and the resumed prepare. Configure it with grpc_options={'grpc_auto_resume_timeout_seconds': 300}. The deadline starts after the suspended response; initial prepare has its own timeout. A dispatched resume may still complete after the client times out.

Set auto_resume=False to disable recovery. Invalid credentials, denied permissions, unsupported servers and failed or unknown cluster states stop recovery without falling back to username/password authentication. Externally supplied access tokens are not refreshed.

Secure Connection Example

To establish a secure connection using SSL/TLS:

conn = Connection(
    host=host,
    port=443,  # Typically 443 for secure connections
    username=username,
    password=password,
    database=database,
    cluster_name='production-cluster',
    secure=True  # Enable SSL/TLS
)

Cluster-Specific Connection

When working with multiple clusters, specify the cluster name:

conn = Connection(
    host=host,
    port=port,
    username=username,
    password=password,
    database=database,
    cluster_name='analytics-cluster-01',  # Specify cluster name
    secure=True
)

HTTPS Connection with HAProxy

When connecting through HAProxy with HTTPS, you can provide a custom CA certificate for secure connections. The ssl_cert parameter accepts either a file path to a PEM certificate or the certificate content as bytes.

Using a CA certificate file path:

conn = Connection(
    host=host,
    port=443,
    username=username,
    password=password,
    database=database,
    secure=True,
    ssl_cert='/path/to/ca-cert.pem'  # Path to your CA certificate
)

Reading certificate content as bytes:

# Read certificate file and pass as bytes
with open('/path/to/ca-cert.pem', 'rb') as cert_file:
    cert_data = cert_file.read()

conn = Connection(
    host=host,
    port=443,
    username=username,
    password=password,
    database=database,
    secure=True,
    ssl_cert=cert_data  # Certificate content as bytes
)

Using system CA bundle for publicly signed certificates:

# When ssl_cert is None, system default CA bundle is used
conn = Connection(
    host=host,
    port=443,
    username=username,
    password=password,
    database=database,
    secure=True  # Uses system CA bundle by default
)

Connection pooling with custom CA certificate:

pool = ConnectionPool(
    min_size=2,
    max_size=10,
    host=host,
    port=443,
    username=username,
    password=password,
    database=database,
    secure=True,
    ssl_cert='/path/to/ca-cert.pem'  # Custom CA certificate for pool connections
)

Handling Missing System Dependencies

The e6data connector uses the fastbinary module (from Apache Thrift) for optimal performance when deserializing data. This module requires system-level dependencies (python3-devel and gcc-c++) to be installed.

Default Behavior (Recommended): By default, the connector requires fastbinary to be available. If it's not found, the connection will fail immediately with a clear error message:

conn = Connection(
    host=host,
    port=port,
    username=username,
    password=password,
    database=database
)
# Raises exception if fastbinary is not available

Fallback to Pure Python: If you cannot install system dependencies (e.g., in restricted environments, serverless platforms, or containers without build tools), you can disable the fastbinary requirement. The connector will fall back to a pure Python implementation with a performance penalty:

conn = Connection(
    host=host,
    port=port,
    username=username,
    password=password,
    database=database,
    require_fastbinary=False  # Allow operation without fastbinary
)
# Logs warning but continues with pure Python implementation

When to use require_fastbinary=False:

• Running in AWS Lambda or other serverless environments
• Docker containers built without compilation tools
• Restricted environments where system packages cannot be installed
• Development/testing environments where performance is not critical

Performance Impact:

• With fastbinary: Optimal performance for data deserialization
• Without fastbinary (pure Python): ~2-3x slower deserialization, but otherwise fully functional

Note: It's strongly recommended to install system dependencies when possible for best performance. The require_fastbinary=False option should only be used when system dependencies cannot be installed.

Perform a Queries & Get Results

query = 'SELECT * FROM <TABLE_NAME>'  # Replace with the query.

cursor = conn.cursor(catalog_name=catalog_name)
query_id = cursor.execute(query)  # The execute function returns a unique query ID, which can be use to abort the query.
all_records = cursor.fetchall()
for row in all_records:
   print(row)

To fetch all the records:

records = cursor.fetchall()

To fetch one record:

record = cursor.fetchone()

To fetch limited records:

limit = 500
records = cursor.fetchmany(limit)

To fetch all the records in buffer to reduce memory consumption:

records_iterator = cursor.fetchall_buffer()  # Returns generator
for item in records_iterator:
    print(item)

To get the execution plan after query execution:

import json
explain_response = cursor.explain_analyse()
query_planner = json.loads(explain_response.get('planner'))

To abort a running query:

query_id = '<query_id>'  # query id from execute function response.
cursor.cancel(query_id)

To get the status of a running or completed query:

query_id = '<query_id>'  # query id from execute function response.
status_response = cursor.status(query_id)

# Check if query execution is complete
is_complete = status_response.status  # Returns True when query is done, False if still running

# Get the total row count
row_count = status_response.rowCount  # Total number of rows in the result set

print(f"Query complete: {is_complete}")
print(f"Row count: {row_count}")

The status() method is useful for:

• Monitoring long-running queries: Poll the status periodically to check if execution is complete
• Checking row counts: Get the total number of rows without fetching all results
• Query progress tracking: Integrate with monitoring systems or progress bars
• Conditional fetching: Decide whether to fetch results based on completion status

Example - Polling for query completion:

import time

cursor = conn.cursor(catalog_name=catalog_name)
query_id = cursor.execute("SELECT * FROM large_table")

# Poll until query is complete
while True:
    status_response = cursor.status(query_id)
    if status_response.status:
        print(f"Query complete! Total rows: {status_response.rowCount}")
        break
    else:
        print("Query still running...")
        time.sleep(1)  # Wait 1 second before checking again

# Now fetch the results
results = cursor.fetchall()

Switch database in an existing connection:

database = '<new_database_name>'  # Replace with the new database.
cursor = conn.cursor(database, catalog_name)

Opt in to multi-chunk result batches

Set enable_result_batch_v2=True to enable V2 batches, continuous result downloading, and bounded parallel chunk decoding together. It defaults to False. There are no separate public prefetch or decoding switches. arraysize and fetchmany(size) only change the number of rows returned to the application.

The planner must support V2 and have ENABLE_GET_NEXT_RESULT_BATCH_V2 enabled. ENABLE_GET_NEXT_CHUNK_V2 controls a separate executor-to-planner boundary. These flags do not extend the planner's query lifetime. Finishing a large result within 900 seconds still needs a measurement on the target deployment.

The optimized path requires ordinary GIL-enabled CPython 3.11 through 3.13 and an import-safe application entry point. Put application startup inside if __name__ == "__main__":, including when using SQLAlchemy. Interactive, daemon and frozen applications are rejected before query submission. Two workers use the explicit spawn method; the application's global multiprocessing start method is unchanged. An unguarded main script can run its other import side effects, so keep all application startup inside the guard.

Both APIs default to unlimited gRPC send and receive message sizes (-1). Explicit positive byte limits are still honored. For sync, set max_receive_message_length and max_send_message_length in grpc_options. For async, set the receive limit with max_receive_message_bytes or max_receive_message_length in grpc_options; if both are supplied, they must agree. Set max_send_message_length in grpc_options for the send limit. The receive limit applies to each message after transport decompression, not the total query result. Unlimited receiving does not limit client memory use; continuous downloading can retain serialized batches while decoding catches up.

These examples use your existing connection_options, sql, and consume:

from e6data_python_connector import Connection


def main():
    options = {**connection_options, "enable_result_batch_v2": True}
    with Connection(**options) as connection:
        with connection.cursor() as cursor:
            cursor.execute(sql)
            for rows in cursor.fetchall_buffer():
                consume(rows)


if __name__ == "__main__":
    main()

import asyncio
from e6data_python_connector.aio import AsyncConnection


async def main():
    options = {**connection_options, "enable_result_batch_v2": True}
    async with AsyncConnection(**options) as connection:
        async with connection.cursor() as cursor:
            await cursor.execute(sql)
            async for rows in cursor.fetchall_buffer():
                consume(rows)


if __name__ == "__main__":
    asyncio.run(main())

SQLAlchemy accepts the same flag through connect_args. Create and use engines inside the guarded application entry point:

connect_args = {**connection_options, "enable_result_batch_v2": True}
engine = create_engine("e6data://", connect_args=connect_args)
async_engine = create_async_engine("e6data+asyncio://", connect_args=connect_args)

After the first V2 response, a download thread (sync) or task (async) keeps fetching serialized batches until the engine returns end-of-stream. Decoding runs concurrently and takes batches from the queue in order. Downloading does not wait for decoding or application row processing. There is no queue-size limit and no process-wide limit on active downloaders. Result RPCs for the same query never overlap.

Each background RPC gets a fresh configured transport budget when it starts; that deadline never extends after dispatch. An empty nonterminal response uses backoff with a finite no-progress deadline. Completed responses can wait while the application is idle. Public fetch deadlines still cover waiting and decoding, and expiry or cancellation stops the downloader. Async fetchall keeps one deadline for its entire operation.

Two shared worker processes can decode one multi-chunk envelope at a time. Other envelopes use the sequential path. Workers receive chunk bytes and column positions, not connections or credentials. Process startup and copying add cost, so parallel decoding is not a promise of better performance for every result.

Fetch return shapes and row order stay the same. fetchall_buffer() yields the original decoded chunks. All chunks in an envelope must decode successfully before any are exposed. The serialized queue can hold the entire result when downloading is faster than decoding or application reads. Workers add Python objects and process copies; the receive limit is not a limit on total memory. fetchall() still retains the full result.

If a decoder worker exits or its local pipe breaks, the connector stops the failed workers and decodes that complete serialized envelope once in the parent process. It uses the same deadline and makes no extra result RPC. Continuous downloading stays active and uncapped during this recovery. Later envelopes use sequential decoding until the decoder runtime is replaced through its normal lifecycle. Cancellation, expired deadlines, invalid worker replies, and errors reported while decoding data are still failures. Their original cause is kept on the first error delivered to the caller.

Only a V2 UNIMPLEMENTED response switches that query to V1 and disables its optimizations. Other RPC failures reach the caller as the original grpc.RpcError or grpc.aio.AioRpcError. A failed pending RPC is raised when its envelope is needed, after current valid chunks have been consumed. Failed or cancelled queries are not replayed. Later fetches on an incomplete result raise IncompleteResultError. Clear, cancel, close and pool return retire pending work; unsafe connections must not be reused.

Use debug=True for payload-free RPC, decode and worker diagnostics. Serialized Protobuf bytes are uncompressed payload size, not bytes sent over the network.

For rollback, close old cursors and connections, then omit enable_result_batch_v2 or set it to False on new connections. Closing the last V2 connection also stops its shared workers. See the qualification instructions. Local synthetic results do not prove customer performance or the 900-second query requirement.

Get Query Time Metrics

import json
query = 'SELECT * FROM <TABLE_NAME>'

cursor = conn.cursor(catalog_name)
query_id = cursor.execute(query)  # execute function returns query id, can be use for aborting the query.
all_records = cursor.fetchall()
explain_response = cursor.explain_analyse()
query_planner = json.loads(explain_response.get('planner'))

execution_time = query_planner.get("total_query_time")  # In milliseconds
queue_time = query_planner.get("executionQueueingTime")  # In milliseconds
parsing_time = query_planner.get("parsingTime")  # In milliseconds
row_count = query_planner.rowcount

Get Schema - a list of Databases, Tables or Columns

The following code returns a dictionary of all databases, all tables and all columns connected to the cluster currently in use. This function can be used without passing database name to get list of all databases.

databases = conn.get_schema_names()  # To get list of databases.
print(databases)

database = '<database_name>'  # Replace with actual database name.
tables = conn.get_tables(database=database)  # To get list of tables from a database.
print(tables)

table_name = '<table_name>'  # Replace with actual table name.
columns = conn.get_tables(database=database, table=table_name)  # To get the list of columns from a table.
columns_with_type = list()
"""
Getting the column name and type.
"""
for column in columns:
   columns_with_type.append(dict(column_name=column.fieldName, column_type=column.fieldType))
print(columns_with_type)

Code Hygiene

It is recommended to clear the cursor, close the cursor and close the connection after running a function as a best practice. This enhances performance by clearing old data from memory.

cursor.clear() # Not needed when aborting a query
cursor.close()
conn.close()

Code Example

The following code is an example which combines a few functions described above.

from e6data_python_connector import Connection
import json

username = '<username>'  # Your e6data Email ID.
password = '<password>'  # Access Token generated in the e6data console.

host = '<host>'  # IP address or hostname of the cluster to be used.
database = '<database>'  # # Database to perform the query on.
port = 80  # Port of the e6data engine.

sql_query = 'SELECT * FROM <TABLE_NAME>'  # Replace with the actual query.

catalog_name = '<catalog_name>'  # Replace with the actual catalog name.

conn = Connection(
    host=host,
    port=port,
    username=username,
    database=database,
    password=password
)

cursor = conn.cursor(db_name=database, catalog_name=catalog_name)
query_id = cursor.execute(sql_query)
all_records = cursor.fetchall()
explain_response = cursor.explain_analyse()
planner_result = json.loads(explain_response.get('planner'))
execution_time = planner_result.get("total_query_time") / 1000  # Converting into seconds.
row_count = cursor.rowcount
columns = [col[0] for col in cursor.description]  # Get the column names and merge them with the results.
results = []
for row in all_records:
   row = dict(zip(columns, row))
   results.append(row)
   print(row)
print('Total row count {}, Execution Time (seconds): {}'.format(row_count, execution_time))
cursor.clear()
cursor.close()
conn.close()

Zero Downtime Deployment

🚀 Zero Downtime Features

The e6data Python Connector provides automatic zero downtime deployment support through intelligent blue-green deployment strategy management:

✅ No Code Changes Required

Your existing applications automatically benefit from zero downtime deployment without any modifications:

# Your existing code works exactly the same
from e6data_python_connector import Connection

conn = Connection(
    host='your-host',
    port=80,
    username='your-email',
    password='your-token',
    database='your-database'
)

cursor = conn.cursor()
cursor.execute("SELECT * FROM your_table")
results = cursor.fetchall()

🔄 Automatic Strategy Detection

• Detects active deployment strategy (blue/green) on connection
• Caches strategy information for optimal performance
• Automatically switches strategies when deployments occur

🛡️ Seamless Query Protection

• Running queries continue uninterrupted during deployments
• New queries automatically use the new deployment strategy
• Graceful transitions ensure no query loss or failures

⚡ Performance Optimized

• < 100ms additional latency on first connection (one-time cost)
• 0ms overhead for 95% of queries (cached strategy)
• < 1KB additional memory usage per connection

🔧 Thread & Process Safe

• Full support for multi-threaded applications
• Process-safe shared memory management
• Concurrent query execution without conflicts

Advanced Configuration (Optional)

For enhanced monitoring and performance tuning:

# Enhanced gRPC configuration for zero downtime
grpc_options = {
    'keepalive_timeout_ms': 60000,      # 1 minute keepalive timeout
    'keepalive_time_ms': 30000,         # 30 seconds keepalive interval
    'max_receive_message_length': 100 * 1024 * 1024,  # 100MB
    'max_send_message_length': 100 * 1024 * 1024,     # 100MB
}

conn = Connection(
    host='your-host',
    port=80,
    username='your-email',
    password='your-token',
    database='your-database',
    grpc_options=grpc_options
)

Environment Configuration

Configure zero downtime features using environment variables:

# Strategy cache timeout (default: 300 seconds)
export E6DATA_STRATEGY_CACHE_TIMEOUT=300

# Maximum retry attempts (default: 5)
export E6DATA_MAX_RETRY_ATTEMPTS=5

# Enable debug logging for strategy operations
export E6DATA_STRATEGY_LOG_LEVEL=INFO

Testing Zero Downtime

Use the included mock server for testing and development:

# Terminal 1: Start mock server
python mock_grpc_server.py

# Terminal 2: Run test client
python test_mock_server.py

# Or use the convenience script
./run_mock_test.sh

📚 Comprehensive Documentation

Explore detailed documentation in the docs/zero-downtime/ directory:

• 📋 Overview - Complete guide and feature overview
• 🔧 API Reference - Detailed API documentation
• 🌊 Flow Documentation - Process flows and diagrams
• 💼 Business Logic - Business rules and decisions
• 🏗️ Architecture - System architecture and design
• ⚙️ Configuration - Complete configuration guide
• 🧪 Testing - Testing strategies and tools
• 🔍 Troubleshooting - Common issues and solutions
• 🚀 Migration Guide - Step-by-step migration instructions

Key Benefits

Migration

Existing applications automatically benefit from zero downtime deployment:

1. Update connector: pip install --upgrade e6data-python-connector
2. No code changes: Your existing code works without modifications
3. Monitor: Use enhanced logging to monitor strategy transitions
4. Validate: Test with your existing applications

For detailed migration instructions, see the Migration Guide.

Performance Optimization

Memory Efficiency

• Use fetchall_buffer() for memory-efficient large result sets
• Automatic cleanup of query-strategy mappings
• Bounded memory usage with TTL-based caching

Network Performance

• Configure gRPC options for optimal network performance
• Intelligent keepalive settings for connection stability
• Message size optimization for large queries

Connection Management

Connection Pooling

The e6data Python connector now includes a built-in connection pool for efficient connection management and reuse across multiple threads. The ConnectionPool class provides:

• Thread-safe connection reuse: Each thread automatically reuses its assigned connection
• Automatic lifecycle management: Handles connection creation, health checks, and cleanup
• Overflow connections: Creates temporary connections when pool is exhausted
• Connection health monitoring: Automatic detection and replacement of broken connections
• Statistics tracking: Monitor pool usage and performance

Basic Connection Pool Usage

from e6data_python_connector import ConnectionPool

# Create a connection pool
pool = ConnectionPool(
    min_size=2,        # Minimum connections to maintain
    max_size=10,       # Maximum connections in pool
    max_overflow=5,    # Additional temporary connections allowed
    timeout=30.0,      # Timeout for getting connection (seconds)
    recycle=3600,      # Maximum age before recycling (seconds)
    debug=False,       # Enable debug logging
    pre_ping=True,     # Check connection health before use
    # Connection parameters
    host=host,
    port=port,
    username=username,
    password=password,
    database=database,
    catalog=catalog_name,
    cluster_name=cluster_name,
    secure=True
)

# Get connection and execute query
conn = pool.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM table")
results = cursor.fetchall()

# Return connection to pool (important!)
pool.return_connection(conn)

# Clean up when done
pool.close_all()

Using Context Manager (Recommended)

The context manager pattern ensures connections are automatically returned to the pool:

from e6data_python_connector import ConnectionPool

pool = ConnectionPool(
    min_size=2,
    max_size=10,
    host=host,
    port=port,
    username=username,
    password=password,
    database=database
)

# Connection automatically returned to pool after use
with pool.get_connection_context() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM table")
    results = cursor.fetchall()
    print(results)

Concurrent Query Execution

Connection pooling is especially beneficial for concurrent query execution:

import concurrent.futures
from e6data_python_connector import ConnectionPool

def execute_query(pool, query_id, query):
    """Execute a query using a pooled connection."""
    # Each thread will reuse its assigned connection
    conn = pool.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(query)
        results = cursor.fetchall()
        return f"Query {query_id}: {len(results)} rows"
    finally:
        pool.return_connection(conn)

# Create pool
pool = ConnectionPool(
    min_size=3,
    max_size=10,
    host=host,
    port=port,
    username=username,
    password=password,
    database=database
)

# Execute multiple queries concurrently
queries = [
    "SELECT COUNT(*) FROM table1",
    "SELECT AVG(value) FROM table2",
    "SELECT MAX(date) FROM table3"
]

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [
        executor.submit(execute_query, pool, i, query)
        for i, query in enumerate(queries)
    ]
    
    for future in concurrent.futures.as_completed(futures):
        print(future.result())

# Clean up
pool.close_all()

Connection Pool Configuration

Monitoring Pool Statistics

# Get pool statistics
stats = pool.get_statistics()
print(f"Active connections: {stats['active_connections']}")
print(f"Idle connections: {stats['idle_connections']}")
print(f"Total requests: {stats['total_requests']}")
print(f"Failed connections: {stats['failed_connections']}")

When to Use Connection Pooling

Connection pooling is recommended when:

• Executing multiple queries concurrently
• Building web applications or APIs
• Running batch processing jobs
• Reducing connection overhead
• Improving application performance

Direct Connection Usage (Without Pool)

For simple, single-threaded applications, you can still use direct connections:

from e6data_python_connector import Connection

conn = Connection(
    host=host,
    port=port,
    username=username,
    password=password,
    database=database
)

cursor = conn.cursor()
cursor.execute("SELECT * FROM table")
results = cursor.fetchall()
conn.close()

Additional Connection Management Features

• Automatic connection health monitoring
• Graceful connection recovery and retry logic
• Blue-green deployment support with automatic failover

Debugging and Troubleshooting

Enable Debug Mode

Enable comprehensive debugging to troubleshoot connection and query issues:

from e6data_python_connector import Connection

conn = Connection(
    host=host,
    port=port,
    username=username,
    password=password,
    database=database,
    debug=True  # Enable debug logging
)

When debug=True, the following features are enabled:

• Connector logging at DEBUG level, for both sync and async connections. Existing application logging handlers and the root logger level are kept.
• Blue-green strategy transition logging
• Connection lifecycle logging
• Query execution detailed logging

Each result fetch writes a normal debug message with protocol, rpc_seconds, decode_seconds, chunk_count, serialized_bytes, and status. The timing values are seconds. rpc_seconds measures the client RPC call, including waiting for its response. decode_seconds measures client response processing. serialized_bytes is the protobuf response size, not the number of bytes on the network. A failed attempt reports zero response size and chunk count. The same values remain available as result_batch_* attributes for structured log handlers.

These metrics do not add SQL, credentials, session IDs, result values, or raw metadata to logs. debug=False keeps the normal quiet default. Applications can also enable DEBUG on the e6data_python_connector logger themselves. debug=True adds a connector stream handler only when no existing handler in its logger path accepts DEBUG. As with normal Python logging, enabling the package logger also affects other open connector connections in that process.

gRPC Network Tracing

For low-level gRPC network debugging (HTTP/2 frames, TCP events), set environment variables before running your Python script:

# Enable gRPC network tracing
export GRPC_VERBOSITY=DEBUG
export GRPC_TRACE=client_channel,http2

# For comprehensive tracing
export GRPC_TRACE=api,call_error,channel,client_channel,connectivity_state,http,http2_stream,tcp,transport_security

# Run your script
python your_script.py

Note: These environment variables must be set before Python starts, as the gRPC C++ core reads them at module import time.

Common Issues and Solutions

See TECH_DOC.md for detailed technical documentation.