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

DB-API Interface

Name: tessl/pypi-pymssql
Author: tessl

Standard Python DB-API 2.0 compliant interface providing Connection and Cursor objects with full transaction support, parameter binding, and result set iteration. This is the recommended interface for most Python applications accessing SQL Server.

Capabilities

Connection Management

Creates and manages database connections with comprehensive parameter support and context manager integration.

def connect(
    server='.', 
    user=None, 
    password=None, 
    database='', 
    timeout=0, 
    login_timeout=60, 
    charset='UTF-8', 
    as_dict=False, 
    host='', 
    appname=None, 
    port='1433', 
    encryption=None,
    read_only=False,
    conn_properties=None, 
    autocommit=False, 
    tds_version=None, 
    use_datetime2=False, 
    arraysize=1
) -> Connection:
    """
    Create a connection to SQL Server.
    
    Parameters:
    - server (str): Database host, default '.'
    - 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 set, default 'UTF-8'
    - as_dict (bool): Return rows as dictionaries instead of tuples
    - host (str): Database host and instance specification (alias for server)
    - appname (str): Application name for connection
    - port (str): TCP port, default '1433'
    - encryption (str): Encryption mode ('off', 'request', 'require')
    - read_only (bool): Connect in read-only mode
    - conn_properties (str|list): SQL queries to run on connection
    - autocommit (bool): Enable autocommit mode
    - tds_version (str): TDS protocol version
    - use_datetime2 (bool): Use DATETIME2 compatibility
    - arraysize (int): Default cursor arraysize
    
    Returns:
    Connection object
    """

Usage example:

# Basic connection
conn = pymssql.connect('localhost', 'sa', 'password', 'mydb')

# Advanced connection with SSL and custom settings
conn = pymssql.connect(
    server='myserver.database.windows.net',
    user='username@myserver',
    password='password',
    database='mydatabase',
    as_dict=True,
    autocommit=False,
    tds_version='7.3',
    conn_properties=[
        'SET TEXTSIZE 2147483647',
        'SET ARITHABORT ON'
    ]
)

Connection Object

Represents an active database connection with transaction control and cursor creation.

class Connection:
    """Database connection object."""
    
    # Properties
    messages: list  # Server messages
    timeout: int    # Query timeout in seconds
    bulk_copy_executemany_page_size: int  # Bulk copy page size
    arraysize: int  # Default cursor array size
    
    def cursor(self, as_dict=None, arraysize=None) -> Cursor:
        """
        Create a cursor object for executing queries.
        
        Parameters:
        - as_dict (bool): Override connection's as_dict setting
        - arraysize (int): Override connection's arraysize setting
        
        Returns:
        Cursor object
        """
    
    def commit(self) -> None:
        """Commit the current transaction."""
    
    def rollback(self) -> None:
        """Roll back the current transaction."""
    
    def close(self) -> None:
        """Close the connection."""
    
    def autocommit(self, status: bool) -> None:
        """
        Set autocommit mode.
        
        Parameters:
        - status (bool): True to enable autocommit, False to disable
        """
    
    def bulk_copy(
        self, 
        table_name: str, 
        elements: list, 
        column_ids=None, 
        batch_size=1000, 
        tablock=False, 
        check_constraints=False, 
        fire_triggers=False
    ) -> None:
        """
        Bulk copy data to a table.
        
        Parameters:
        - table_name (str): Target table name
        - elements (list): List of tuples containing row data
        - column_ids (list): Target column IDs (1-based), default all columns
        - batch_size (int): Rows per batch, default 1000
        - tablock (bool): Set TABLOCK hint
        - check_constraints (bool): Set CHECK_CONSTRAINTS hint  
        - fire_triggers (bool): Set FIRE_TRIGGERS hint
        """
    
    def __enter__(self) -> 'Connection':
        """Context manager entry."""
        
    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """Context manager exit with automatic cleanup."""

Cursor Object

Executes SQL statements and fetches results with support for parameter binding and multiple result sets.

class Cursor:
    """Database cursor for executing queries."""
    
    # Properties
    connection: Connection      # Parent connection
    description: tuple         # Column descriptions
    rowcount: int             # Rows affected by last operation
    rownumber: int            # Current row position (0-based)
    lastrowid: int            # Last inserted identity value
    returnvalue: int          # Stored procedure return value
    arraysize: int            # Fetch array size
    
    def execute(self, operation: str, params=None) -> None:
        """
        Execute a SQL statement.
        
        Parameters:
        - operation (str): SQL statement with %s or %(name)s placeholders
        - params: Parameter values (tuple, dict, or single value)
        """
    
    def executemany(self, operation: str, seq_of_parameters, *, batch_size=-1) -> None:
        """
        Execute SQL statement multiple times.
        
        Parameters:
        - operation (str): SQL statement
        - seq_of_parameters: Sequence of parameter sets
        - batch_size (int): Batch size, -1 uses arraysize
        """
    
    def fetchone(self):
        """
        Fetch next row.
        
        Returns:
        Row as tuple or dict (based on as_dict), None if no more rows
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows.
        
        Parameters:
        - size (int): Number of rows to fetch, None uses arraysize
        
        Returns:
        List of rows as tuples or dicts
        """
    
    def fetchall(self):
        """
        Fetch all remaining rows.
        
        Returns:
        List of rows as tuples or dicts
        """
    
    def nextset(self) -> bool:
        """
        Move to next result set.
        
        Returns:
        True if next result set available, None otherwise
        """
    
    def callproc(self, procname: str, parameters=()) -> tuple:
        """
        Call a stored procedure.
        
        Parameters:
        - procname (str): Stored procedure name
        - parameters (tuple): Input parameters
        
        Returns:
        Tuple of parameter values (input and output)
        """
    
    def close(self) -> None:
        """Close the cursor."""
    
    def setinputsizes(self, sizes=None) -> None:
        """Set input parameter sizes (no-op in pymssql)."""
    
    def setoutputsize(self, size=None, column=0) -> None:
        """Set output column size (no-op in pymssql)."""
    
    def __iter__(self):
        """Iterator protocol support."""
    
    def __next__(self):
        """Get next row in iteration."""
    
    def __enter__(self) -> 'Cursor':
        """Context manager entry."""
        
    def __exit__(self, exc_type, exc_value, traceback) -> None:
        """Context manager exit with automatic cleanup."""

Transaction Management

Example of transaction handling:

conn = pymssql.connect('server', 'user', 'password', 'database')

try:
    cursor = conn.cursor()
    
    # Begin transaction (implicit)
    cursor.execute("INSERT INTO users (name) VALUES (%s)", ('Alice',))
    cursor.execute("INSERT INTO orders (user_id, amount) VALUES (%s, %s)", (1, 100.00))
    
    # Commit transaction
    conn.commit()
    print("Transaction committed successfully")
    
except Exception as e:
    # Rollback on error
    conn.rollback()
    print(f"Transaction rolled back: {e}")
    
finally:
    conn.close()

Context Manager Usage

# Connection context manager
with pymssql.connect('server', 'user', 'password', 'database') as conn:
    with conn.cursor(as_dict=True) as cursor:
        cursor.execute("SELECT * FROM users WHERE active = %s", (True,))
        for row in cursor:
            print(f"User: {row['name']}")
    # Cursor automatically closed
# Connection automatically closed

Data Types

DB-API Type Objects

class DBAPIType:
    """DB-API type object for type comparison."""
    
    def __init__(self, value: int): ...
    def __eq__(self, other) -> bool: ...
    
# Type constants
STRING: DBAPIType    # String types
BINARY: DBAPIType    # Binary types  
NUMBER: DBAPIType    # Numeric types
DATETIME: DBAPIType  # Date/time types
DECIMAL: DBAPIType   # Decimal types

Date/Time Support

# Standard datetime aliases
Date = datetime.date
Time = datetime.time  
Timestamp = datetime.datetime

# Helper object for parameterless queries
NoParams: object

Install with Tessl CLI