tessl/pypi-pymssql

DB-API interface to Microsoft SQL Server for Python with comprehensive SQL Server integration and FreeTDS support.

—

Pending

Overview

Eval results

Files

Connection Configuration

Name: tessl/pypi-pymssql
Author: tessl

Comprehensive connection management with support for various authentication methods, SSL/TLS encryption, connection pooling, and SQL Server-specific features. Covers both high-level and low-level connection configuration options.

Capabilities

Basic Connection Parameters

Core connection parameters supported by both pymssql.connect() and _mssql.connect().

def connect(
    server='.', 
    user=None, 
    password=None, 
    database='', 
    timeout=0, 
    login_timeout=60, 
    charset='UTF-8', 
    appname=None, 
    port='1433'
) -> Connection:
    """
    Basic connection parameters.
    
    Parameters:
    - server (str): Database server and instance, default '.' (local)
    - user (str): Database username
    - password (str): User password  
    - database (str): Database name to connect to
    - timeout (int): Query timeout in seconds, 0 = no timeout
    - login_timeout (int): Connection timeout in seconds, default 60
    - charset (str): Character encoding, default 'UTF-8'
    - appname (str): Application name shown in SQL Server
    - port (str): TCP port number, default '1433'
    """

Advanced Connection Options

Extended configuration options for specialized use cases.

def connect(
    # ... basic parameters ...
    as_dict=False,
    autocommit=False, 
    tds_version=None,
    use_datetime2=False,
    arraysize=1,
    conn_properties=None,
    host='',
    encryption=None,
    read_only=False
) -> Connection:
    """
    Advanced connection parameters.
    
    Parameters:
    - as_dict (bool): Return rows as dictionaries instead of tuples
    - autocommit (bool): Enable autocommit mode
    - tds_version (str): TDS protocol version ('7.0', '7.1', '7.2', '7.3')
    - use_datetime2 (bool): Use DATETIME2 compatible conversion
    - arraysize (int): Default cursor arraysize for fetchmany()
    - conn_properties (str|list): SQL commands to execute on connection
    - host (str): Alternative server specification format
    - encryption (str): SSL/TLS mode ('off', 'request', 'require')
    - read_only (bool): Connect in read-only mode (SQL Server 2012+)
    """

Connection Properties

SQL commands executed automatically upon connection establishment.

# Default connection properties (applied automatically)
DEFAULT_CONN_PROPERTIES = [
    'SET ARITHABORT ON',
    'SET CONCAT_NULL_YIELDS_NULL ON', 
    'SET ANSI_NULLS ON',
    'SET ANSI_NULL_DFLT_ON ON',
    'SET ANSI_PADDING ON',
    'SET ANSI_WARNINGS ON',
    'SET CURSOR_CLOSE_ON_COMMIT ON',
    'SET QUOTED_IDENTIFIER ON',
    'SET TEXTSIZE 2147483647'
]

Global Configuration

Module-level functions for connection management.

def get_max_connections() -> int:
    """
    Get maximum number of simultaneous connections allowed.
    
    Returns:
    Maximum connection count (default: 25)
    """

def set_max_connections(number: int) -> None:
    """
    Set maximum number of simultaneous connections allowed.
    
    Parameters:
    - number (int): Maximum connections (minimum: 1)
    """

def get_dbversion() -> str:
    """
    Get FreeTDS library version information.
    
    Returns:
    Version string (e.g., "freetds v1.2.5")
    """

def set_wait_callback(callback) -> None:
    """
    Set callback for cooperative multitasking support.
    
    Parameters:
    - callback: Function with signature callback(read_fileno)
    """

def version_info() -> str:
    """
    Get comprehensive version information for debugging.
    
    Returns:
    Version information string including pymssql, FreeTDS, Python, and OS
    """

Usage Examples

Basic Connections

import pymssql

# Local SQL Server with Windows Authentication
conn = pymssql.connect(server='.', database='mydb')

# Remote SQL Server with SQL Authentication  
conn = pymssql.connect(
    server='192.168.1.100',
    user='myuser',
    password='mypassword', 
    database='mydb'
)

# Named instance on specific port
conn = pymssql.connect(
    server='MYSERVER\\SQLEXPRESS',
    user='sa',
    password='password',
    database='testdb'
)

# TCP/IP with custom port
conn = pymssql.connect(
    server='myserver.company.com,1433',
    user='appuser',
    password='apppass',
    database='proddb'
)

Azure SQL Database

import pymssql

# Azure SQL Database connection
conn = pymssql.connect(
    server='myserver.database.windows.net',
    user='username@myserver',  # Include server name in username
    password='password',
    database='mydatabase',
    encryption='require',      # Force SSL encryption
    tds_version='7.3'         # Use latest TDS version
)

# Azure with connection string format
conn = pymssql.connect(
    host='myserver.database.windows.net',
    user='username',
    password='password', 
    database='mydatabase',
    port='1433',
    encryption='require'
)

High-Performance Configuration

import pymssql

# Optimized for bulk operations
conn = pymssql.connect(
    server='localhost',
    user='sa',
    password='password',
    database='warehouse',
    autocommit=True,           # Reduce transaction overhead
    timeout=300,               # 5 minute query timeout
    arraysize=1000,           # Large fetch size
    conn_properties=[
        'SET ARITHABORT ON',
        'SET ANSI_WARNINGS OFF',  # Reduce warning messages
        'SET TEXTSIZE 2147483647',
        'SET LOCK_TIMEOUT 30000'  # 30 second lock timeout
    ]
)

# Configure bulk copy page size
conn.bulk_copy_executemany_page_size = 10000

Development and Debugging

import pymssql
from pymssql import _mssql

# Enable query debugging
conn = _mssql.connect(
    server='localhost',
    user='dev',
    password='devpass',
    database='testdb'
)
conn.debug_queries = True  # Print all queries to stderr

# Connection with custom application name for monitoring
conn = pymssql.connect(
    server='localhost',
    user='dev',
    password='devpass',
    database='testdb',
    appname='MyApp-Development',
    timeout=10                # Short timeout for development
)

# Custom message handler for server messages
def message_handler(msgstate, severity, srvname, procname, line, msgtext):
    print(f"Server message: {msgtext} (severity: {severity})")

conn_low = _mssql.connect('localhost', 'dev', 'devpass', 'testdb')
conn_low.set_msghandler(message_handler)

Connection Pooling Pattern

import threading
import queue
import pymssql

class ConnectionPool:
    def __init__(self, max_connections=10, **conn_params):
        self.max_connections = max_connections
        self.conn_params = conn_params
        self.pool = queue.Queue(maxsize=max_connections)
        self.lock = threading.Lock()
        
        # Pre-populate pool
        for _ in range(max_connections):
            conn = pymssql.connect(**conn_params)
            self.pool.put(conn)
    
    def get_connection(self):
        return self.pool.get()
    
    def return_connection(self, conn):
        self.pool.put(conn)

# Usage
pool = ConnectionPool(
    max_connections=25,
    server='localhost',
    user='pooluser',
    password='poolpass',
    database='mydb',
    autocommit=True
)

# Use connection from pool
conn = pool.get_connection()
try:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    # ... process results ...
finally:
    pool.return_connection(conn)

TDS Version Configuration

import pymssql

# TDS version selection for compatibility
versions = {
    'legacy': '7.0',    # SQL Server 7.0 / 2000
    'standard': '7.1',  # SQL Server 2000 SP1+
    'modern': '7.2',    # SQL Server 2005+
    'latest': '7.3'     # SQL Server 2008+
}

# Connect with specific TDS version
conn = pymssql.connect(
    server='legacy-server',
    user='user',
    password='pass',
    database='olddb',
    tds_version=versions['legacy']
)

# Check negotiated TDS version
from pymssql import _mssql
conn_low = _mssql.connect('server', 'user', 'pass', 'db', tds_version='7.3')
print(f"TDS version: {conn_low.tds_version}")
print(f"TDS version tuple: {conn_low.tds_version_tuple}")

Connection String Patterns

import pymssql

# Different server specification formats
connections = {
    # Local default instance
    'local_default': pymssql.connect('.', 'sa', 'pass', 'db'),
    
    # Local named instance  
    'local_named': pymssql.connect('.\\SQLEXPRESS', 'sa', 'pass', 'db'),
    
    # Remote with port
    'remote_port': pymssql.connect('server:1433', 'user', 'pass', 'db'),
    
    # Remote with comma syntax
    'remote_comma': pymssql.connect('server,1433', 'user', 'pass', 'db'),
    
    # IP address
    'ip_address': pymssql.connect('192.168.1.100', 'user', 'pass', 'db'),
    
    # FQDN
    'fqdn': pymssql.connect('sql.company.com', 'user', 'pass', 'db')
}

Error Handling and Timeouts

import pymssql
from pymssql.exceptions import *

try:
    # Connection with timeouts
    conn = pymssql.connect(
        server='slow-server',
        user='user',
        password='password',
        database='db',
        login_timeout=30,     # 30 second connection timeout
        timeout=60           # 60 second query timeout
    )
    
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM large_table")
    
except InterfaceError as e:
    print(f"Connection interface error: {e}")
except OperationalError as e:
    print(f"Connection operational error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Configuration Best Practices

Production Environment

import pymssql

# Production configuration
conn = pymssql.connect(
    server='prod-server.company.com',
    user='app_user',            # Dedicated application user
    password='strong_password',  # Strong password
    database='production_db',
    
    # Security
    encryption='require',       # Force SSL
    
    # Performance  
    autocommit=False,          # Explicit transaction control
    timeout=30,                # Reasonable query timeout
    arraysize=100,             # Balanced fetch size
    
    # Monitoring
    appname='ProductionApp-v1.2.3',
    
    # Reliability
    login_timeout=15,          # Quick connection timeout
    conn_properties=[
        'SET ARITHABORT ON',
        'SET LOCK_TIMEOUT 10000'  # 10 second lock timeout
    ]
)

Connection Validation

import pymssql
from pymssql import _mssql

def validate_connection(conn):
    """Validate that connection is healthy."""
    try:
        if hasattr(conn, 'connected'):
            # Low-level connection
            if not conn.connected:
                return False
            conn.execute_scalar("SELECT 1")
        else:
            # High-level connection
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            cursor.fetchone()
            cursor.close()
        return True
    except:
        return False

# Usage
conn = pymssql.connect('server', 'user', 'pass', 'db')
if not validate_connection(conn):
    conn.close()
    conn = pymssql.connect('server', 'user', 'pass', 'db')

Install with Tessl CLI