DB-API 2.0 Interface

def connect(user=None, password=None, host=None, port=None, database=None, **kw):
    """
    Create a DB-API 2.0 compatible database connection.
    
    Parameters:
    - user (str, optional): Database username
    - password (str, optional): Database password  
    - host (str, optional): Database server hostname
    - port (int, optional): Database server port
    - database (str, optional): Database name
    - **kw: Additional connection parameters
    
    Returns:
    Connection: DB-API 2.0 connection object
    
    Raises:
    InterfaceError: If connection parameters are invalid
    OperationalError: If connection cannot be established
    """

class Connection:
    """
    DB-API 2.0 compatible connection interface.
    """
    
    def cursor():
        """
        Create a new cursor object for executing queries.
        
        Returns:
        Cursor: Database cursor for query execution
        """
    
    def commit():
        """
        Commit the current transaction.
        
        Raises:
        DatabaseError: If commit fails
        """
    
    def rollback():
        """
        Roll back the current transaction.
        
        Raises:
        DatabaseError: If rollback fails
        """
    
    def close():
        """
        Close the database connection.
        """
    
    @property
    def autocommit():
        """
        Get/set autocommit mode.
        
        Returns:
        bool: True if autocommit is enabled
        """
    
    def __enter__():
        """Context manager entry."""
    
    def __exit__(exc_type, exc_val, exc_tb):
        """Context manager exit with automatic commit/rollback."""

class Cursor:
    """
    DB-API 2.0 compatible cursor interface for query execution.
    """
    
    def execute(query, parameters=None):
        """
        Execute a query with optional parameters.
        
        Parameters:
        - query (str): SQL query with %(name)s or %s parameter placeholders
        - parameters (dict or sequence, optional): Parameter values
        
        Raises:
        ProgrammingError: If query is invalid
        DataError: If parameters are invalid
        """
    
    def executemany(query, parameter_sequences):
        """
        Execute a query multiple times with different parameter sets.
        
        Parameters:
        - query (str): SQL query with parameter placeholders
        - parameter_sequences (sequence): Sequence of parameter sets
        
        Raises:
        ProgrammingError: If query is invalid
        """
    
    def fetchone():
        """
        Fetch next row from query results.
        
        Returns:
        tuple or None: Next result row as tuple, or None if no more rows
        """
    
    def fetchmany(size=None):
        """
        Fetch multiple rows from query results.
        
        Parameters:
        - size (int, optional): Number of rows to fetch (default: cursor.arraysize)
        
        Returns:
        list: List of result rows as tuples
        """
    
    def fetchall():
        """
        Fetch all remaining rows from query results.
        
        Returns:
        list: All remaining result rows as tuples
        """
    
    def close():
        """Close the cursor."""
    
    def __iter__():
        """Iterator interface for result rows."""
    
    def __next__():
        """Get next result row."""
    
    @property
    def description():
        """
        Get column description for last query.
        
        Returns:
        list: List of 7-tuples describing each column:
              (name, type_code, display_size, internal_size, precision, scale, null_ok)
        """
    
    @property
    def rowcount():
        """
        Get number of rows affected by last query.
        
        Returns:
        int: Row count (-1 if not available)
        """
    
    @property
    def arraysize():
        """
        Get/set default number of rows for fetchmany().
        
        Returns:
        int: Default fetch size
        """

# API compliance constants
apilevel: str      # "2.0"
threadsafety: int  # 1 (threads may share module, not connections)
paramstyle: str    # "pyformat" (%(name)s parameter style)

# Type objects for column type identification
STRING: type       # String data type
BINARY: type       # Binary data type  
NUMBER: type       # Numeric data type
DATETIME: type     # Date/time data type
ROWID: type        # Row ID data type

# Base exceptions
class Warning(Exception):
    """Important warnings raised by database operations."""

class Error(Exception):
    """Base class for all database errors."""

class InterfaceError(Error):
    """Errors in database interface usage."""

class DatabaseError(Error):
    """Errors in database operations."""

# DatabaseError subclasses  
class DataError(DatabaseError):
    """Errors in processed data (invalid values, overflow, etc.)."""

class OperationalError(DatabaseError):
    """Errors in database operations outside user control."""

class IntegrityError(DatabaseError):
    """Database referential integrity violations."""

class InternalError(DatabaseError):
    """Internal database errors."""

class ProgrammingError(DatabaseError):
    """Programming errors (invalid queries, missing tables, etc.)."""

class NotSupportedError(DatabaseError):
    """Unsupported database operations."""

import postgresql.driver.dbapi20 as dbapi

# Connect using DB-API 2.0 interface
conn = dbapi.connect(
    user='postgres',
    password='password',
    host='localhost',
    port=5432,
    database='mydb'
)

# Create cursor for queries
cur = conn.cursor()

# Execute query with parameters
cur.execute("SELECT id, name FROM users WHERE age > %(min_age)s", {'min_age': 18})

# Fetch results
results = cur.fetchall()
for user_id, name in results:
    print(f"User {user_id}: {name}")

# Check query metadata
print("Columns:", [desc[0] for desc in cur.description])
print("Row count:", cur.rowcount)

# Close cursor and connection
cur.close()
conn.close()

import postgresql.driver.dbapi20 as dbapi

conn = dbapi.connect(database='mydb', user='postgres')

try:
    cur = conn.cursor()
    
    # Start transaction (implicit)
    cur.execute("INSERT INTO accounts (name, balance) VALUES (%s, %s)", 
                ("Alice", 1000))
    cur.execute("INSERT INTO accounts (name, balance) VALUES (%s, %s)", 
                ("Bob", 500))
    
    # Transfer money between accounts
    cur.execute("UPDATE accounts SET balance = balance - %s WHERE name = %s", 
                (100, "Alice"))
    cur.execute("UPDATE accounts SET balance = balance + %s WHERE name = %s", 
                (100, "Bob"))
    
    # Commit transaction
    conn.commit()
    print("Transfer completed successfully")
    
except dbapi.DatabaseError as e:
    # Rollback on any error
    conn.rollback()
    print(f"Transfer failed: {e}")
    
finally:
    cur.close()
    conn.close()

import postgresql.driver.dbapi20 as dbapi

# Automatic connection management
with dbapi.connect(database='mydb', user='postgres') as conn:
    with conn.cursor() as cur:
        # Execute multiple queries
        cur.execute("SELECT COUNT(*) FROM users")
        user_count = cur.fetchone()[0]
        
        cur.execute("SELECT COUNT(*) FROM orders") 
        order_count = cur.fetchone()[0]
        
        print(f"Users: {user_count}, Orders: {order_count}")
        
    # Transaction automatically committed if no exceptions
    # Connection automatically closed

import postgresql.driver.dbapi20 as dbapi

conn = dbapi.connect(database='mydb', user='postgres')
cur = conn.cursor()

# Prepare data
user_data = [
    ('Alice', 'alice@example.com', 25),
    ('Bob', 'bob@example.com', 30),
    ('Charlie', 'charlie@example.com', 35)
]

# Execute batch insert
cur.executemany(
    "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
    user_data
)

print(f"Inserted {cur.rowcount} users")

# Fetch inserted users
cur.execute("SELECT * FROM users WHERE name IN %s", (('Alice', 'Bob', 'Charlie'),))

# Process results with iterator
for row in cur:
    user_id, name, email, age = row
    print(f"User {user_id}: {name} ({email}), age {age}")

conn.commit()
cur.close()
conn.close()

import postgresql.driver.dbapi20 as dbapi

try:
    conn = dbapi.connect(database='nonexistent', user='postgres')
except dbapi.OperationalError as e:
    print(f"Connection failed: {e}")
    
try:
    conn = dbapi.connect(database='mydb', user='postgres')
    cur = conn.cursor()
    
    # This will raise ProgrammingError if table doesn't exist
    cur.execute("SELECT * FROM nonexistent_table")
    
except dbapi.ProgrammingError as e:
    print(f"SQL error: {e}")
    
except dbapi.DataError as e:
    print(f"Data error: {e}")
    
except dbapi.IntegrityError as e:
    print(f"Integrity constraint violation: {e}")
    
except dbapi.DatabaseError as e:
    print(f"Database error: {e}")
    
finally:
    if 'cur' in locals():
        cur.close()
    if 'conn' in locals():
        conn.close()

import postgresql.driver.dbapi20 as dbapi

conn = dbapi.connect(database='mydb', user='postgres')
cur = conn.cursor()

# Execute query to get column information
cur.execute("""
    SELECT id, name, email, created_at, is_active, balance
    FROM users 
    LIMIT 1
""")

# Inspect column types
for i, desc in enumerate(cur.description):
    name, type_code, display_size, internal_size, precision, scale, null_ok = desc
    
    # Identify column types
    if type_code == dbapi.STRING:
        type_name = "STRING"
    elif type_code == dbapi.NUMBER:
        type_name = "NUMBER"
    elif type_code == dbapi.DATETIME:
        type_name = "DATETIME"
    else:
        type_name = "OTHER"
    
    print(f"Column {i}: {name} ({type_name})")

cur.close()
conn.close()

# Example showing DB-API 2.0 compatibility with SQLAlchemy
from sqlalchemy import create_engine
import postgresql.driver.dbapi20

# Register py-postgresql as DB-API driver
engine = create_engine(
    'postgresql+pg8000://user:pass@localhost/mydb',
    module=postgresql.driver.dbapi20
)

# Now SQLAlchemy can use py-postgresql through DB-API 2.0 interface
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM users")
    for row in result:
        print(row)