Low-Level Operations

def connect(
    server='.', 
    user=None, 
    password=None, 
    database='', 
    charset='UTF-8', 
    appname=None, 
    port='1433', 
    encryption=None,
    read_only=False,
    tds_version=None, 
    conn_properties=None,
    use_datetime2=False
) -> MSSQLConnection:
    """
    Create a low-level connection to SQL Server.
    
    Parameters:
    - server (str): Database server and instance
    - user (str): Database username  
    - password (str): User password
    - database (str): Database name
    - charset (str): Character set, default 'UTF-8'
    - appname (str): Application name
    - port (str): TCP port, default '1433'
    - encryption (str): Encryption mode ('off', 'request', 'require')
    - read_only (bool): Connect in read-only mode
    - tds_version (str): TDS protocol version
    - conn_properties (str|list): SQL queries to run on connection
    - use_datetime2 (bool): Use DATETIME2 compatibility
    
    Returns:
    MSSQLConnection object
    """

class MSSQLConnection:
    """Low-level SQL Server connection."""
    
    # Properties
    connected: bool              # Connection status
    charset: str                # Character set name
    identity: int               # Last inserted identity value
    query_timeout: int          # Query timeout in seconds
    rows_affected: int          # Rows affected by last query
    debug_queries: bool         # Enable query debugging
    tds_version: str           # TDS version string (e.g., "7.3")
    tds_version_tuple: tuple   # TDS version as tuple (e.g., (7, 3))
    
    def execute_query(self, query_string: str, params=None) -> None:
        """
        Execute a query that returns results.
        
        Parameters:
        - query_string (str): SQL query with %s placeholders
        - params: Parameter values (tuple, dict, or single value)
        
        Use with iteration to get results:
        conn.execute_query("SELECT * FROM users WHERE age > %s", (25,))
        for row in conn:
            print(row)
        """
    
    def execute_non_query(self, query_string: str, params=None) -> None:
        """
        Execute a query that doesn't return results (INSERT, UPDATE, DELETE).
        
        Parameters:
        - query_string (str): SQL query with %s placeholders
        - params: Parameter values (tuple, dict, or single value)
        """
    
    def execute_scalar(self, query_string: str, params=None):
        """
        Execute a query and return the first column of the first row.
        
        Parameters:
        - query_string (str): SQL query with %s placeholders
        - params: Parameter values (tuple, dict, or single value)
        
        Returns:
        Single value from first column of first row
        """
    
    def execute_row(self, query_string: str, params=None) -> tuple:
        """
        Execute a query and return the first row.
        
        Parameters:
        - query_string (str): SQL query with %s placeholders
        - params: Parameter values (tuple, dict, or single value)
        
        Returns:
        First row as tuple
        """
    
    def nextresult(self) -> bool:
        """
        Move to next result set.
        
        Returns:
        True if next result set available, None otherwise
        """
    
    def cancel(self) -> None:
        """Cancel all pending results."""
    
    def close(self) -> None:
        """Close the connection."""
    
    def select_db(self, dbname: str) -> None:
        """
        Change the current database.
        
        Parameters:
        - dbname (str): Database name to switch to
        """
    
    def get_header(self) -> list:
        """
        Get column header information for current result set.
        
        Returns:
        List of 7-element tuples with column descriptions
        """
    
    def init_procedure(self, name: str) -> 'MSSQLStoredProcedure':
        """
        Create a stored procedure object.
        
        Parameters:
        - name (str): Stored procedure name
        
        Returns:
        MSSQLStoredProcedure object
        """
    
    def set_msghandler(self, handler) -> None:
        """
        Set message handler function for server messages.
        
        Parameters:
        - handler: Function with signature (msgstate, severity, srvname, procname, line, msgtext)
        """
    
    def __iter__(self):
        """Iterator protocol for result rows."""
    
    def __next__(self) -> tuple:
        """Get next row from current result set."""

class MSSQLStoredProcedure:
    """Stored procedure execution object."""
    
    # Properties
    connection: MSSQLConnection  # Parent connection
    name: str                   # Procedure name
    parameters: list            # Bound parameters
    
    def bind(
        self, 
        value, 
        dbtype: int, 
        name=None, 
        output=False, 
        null=False, 
        max_length=-1
    ) -> None:
        """
        Bind a parameter to the stored procedure.
        
        Parameters:
        - value: Parameter value
        - dbtype (int): SQL Server data type constant (SQLINT4, SQLVARCHAR, etc.)
        - name (str): Parameter name in "@name" format
        - output (bool): True for output parameters
        - null (bool): True to bind NULL value
        - max_length (int): Maximum length for output parameters
        """
    
    def execute(self) -> None:
        """Execute the stored procedure."""

class MSSQLRowIterator:
    """Iterator for query results."""
    
    def __init__(self, connection: MSSQLConnection, row_format: int): ...
    def __iter__(self): ...
    def __next__(self) -> tuple: ...

# Row format constants
ROW_FORMAT_TUPLE: int = 1
ROW_FORMAT_DICT: int = 2

import _mssql

# Connect to database
conn = _mssql.connect('localhost', 'sa', 'password', 'testdb')

# Execute query with results
conn.execute_query("SELECT name, age FROM users WHERE active = %s", (True,))
for row in conn:
    print(f"Name: {row[0]}, Age: {row[1]}")

# Execute scalar query
count = conn.execute_scalar("SELECT COUNT(*) FROM users")
print(f"Total users: {count}")

# Execute single row query
user = conn.execute_row("SELECT * FROM users WHERE id = %s", (1,))
print(f"User data: {user}")

# Execute non-query
conn.execute_non_query("INSERT INTO users (name, age) VALUES (%s, %s)", ('Alice', 25))
print(f"Rows affected: {conn.rows_affected}")

conn.close()

import _mssql

conn = _mssql.connect('localhost', 'sa', 'password', 'testdb')

# Execute query that returns multiple result sets
conn.execute_query("""
    SELECT * FROM users;
    SELECT * FROM orders;
""")

# Process first result set
print("Users:")
for row in conn:
    print(row)

# Move to next result set
if conn.nextresult():
    print("Orders:")
    for row in conn:
        print(row)

conn.close()

import _mssql

conn = _mssql.connect('localhost', 'sa', 'password', 'testdb')

# Create stored procedure object
proc = conn.init_procedure('GetUsersByAge')

# Bind input parameter
proc.bind(25, _mssql.SQLINT4, '@MinAge')

# Bind output parameter  
proc.bind(0, _mssql.SQLINT4, '@UserCount', output=True)

# Execute procedure
proc.execute()

# Get results
for row in conn:
    print(f"User: {row}")

# Check output parameter value
print(f"Output parameters: {proc.parameters}")

conn.close()

import _mssql

# Advanced connection with custom properties
conn = _mssql.connect(
    server='myserver.database.windows.net',
    user='username@myserver', 
    password='password',
    database='mydatabase',
    charset='UTF-8',
    appname='MyPythonApp',
    tds_version='7.3',
    conn_properties=[
        'SET TEXTSIZE 2147483647',
        'SET ARITHABORT ON',
        'SET ANSI_NULLS ON'
    ]
)

# Set query timeout
conn.query_timeout = 30

# Enable query debugging
conn.debug_queries = True

# Execute query
conn.execute_query("SELECT @@VERSION")
for row in conn:
    print(f"SQL Server version: {row[0]}")

conn.close()

# Integer types
SQLBIT: int
SQLINT1: int      # TINYINT
SQLINT2: int      # SMALLINT  
SQLINT4: int      # INT
SQLINT8: int      # BIGINT
SQLINTN: int      # Variable length integer

# Floating point types
SQLFLT4: int      # REAL
SQLFLT8: int      # FLOAT
SQLFLTN: int      # Variable length float

# Decimal types
SQLDECIMAL: int   # DECIMAL
SQLNUMERIC: int   # NUMERIC

# Money types
SQLMONEY: int     # MONEY
SQLMONEY4: int    # SMALLMONEY
SQLMONEYN: int    # Variable length money

# Character types
SQLCHAR: int      # CHAR
SQLVARCHAR: int   # VARCHAR
SQLTEXT: int      # TEXT

# Binary types  
SQLBINARY: int    # BINARY
SQLVARBINARY: int # VARBINARY
SQLIMAGE: int     # IMAGE

# Date/time types
SQLDATETIME: int  # DATETIME
SQLDATETIM4: int  # SMALLDATETIME
SQLDATETIMN: int  # Variable length datetime

# Other types
SQLBIT: int       # BIT
SQLBITN: int      # Variable length bit
SQLUUID: int      # UNIQUEIDENTIFIER

def quote_simple_value(value, use_datetime2=False, charset='utf-8') -> str:
    """
    Quote a single value for SQL insertion.
    
    Parameters:
    - value: Value to quote
    - use_datetime2 (bool): Use DATETIME2 formatting
    - charset (str): Character encoding
    
    Returns:
    Quoted SQL string
    """

def quote_data(data, use_datetime2=False, charset='utf-8') -> str:
    """
    Quote a data structure for SQL insertion.
    
    Parameters:
    - data: Data structure to quote (dict, tuple, list)
    - use_datetime2 (bool): Use DATETIME2 formatting
    - charset (str): Character encoding
    
    Returns:
    Quoted SQL string
    """

def substitute_params(toformat: str, params=None, use_datetime2=False, charset='utf-8') -> str:
    """
    Substitute parameters in SQL string.
    
    Parameters:
    - toformat (str): SQL string with %s placeholders
    - params: Parameter values
    - use_datetime2 (bool): Use DATETIME2 formatting
    - charset (str): Character encoding
    
    Returns:
    SQL string with substituted parameters
    """

def remove_locale(value: bytes) -> bytes:
    """
    Remove locale-specific formatting from byte string.
    
    Parameters:
    - value (bytes): Input byte string
    
    Returns:
    Byte string with locale formatting removed
    """