[Python] Manipulating Various Databases: List of Common DB-API Methods

2025年12月31日

In Python, when working with different database systems such as SQLite, MySQL, PostgreSQL, or Oracle, we use drivers implemented based on a common specification called DB-API 2.0 (PEP 249).

This allows us to unify the Python-side code (procedures for connection, execution, retrieval, commit, etc.) with almost the same syntax, even if the database product changes.

Here, I have summarized the main operation methods in a table and introduced a standard implementation example using the MySQL driver (mysqlclient / MySQLdb).

Connection-Related Operations

These are the methods for establishing a connection to the database and managing transactions.

Target	Method/Function	Description
Module	`connect(...)`	Establishes a connection to the database and returns a connection object. Arguments specify the host name, user name, password, etc.
Connection	`conn.cursor()`	Creates and returns a cursor object for executing SQL commands.
Connection	`conn.commit()`	Confirms the transaction and applies pending changes (INSERT, UPDATE, DELETE, etc.) to the database.
Connection	`conn.rollback()`	Cancels the transaction, discarding changes made since the last commit and reverting to the previous state. Used when an error occurs.
Connection	`conn.close()`	Closes the connection to the database. Be sure to call this after processing is finished to release resources.

Cursor (Execution) Related Operations

These are the methods for executing SQL and retrieving results.

Target	Method	Description
Cursor	`cur.execute(sql, params)`	Executes an SQL statement. Using placeholders helps prevent SQL injection.
Cursor	`cur.fetchone()`	Retrieves the next row from the query execution result. Returns `None` if there is no data.
Cursor	`cur.fetchall()`	Retrieves all remaining rows from the query execution result as a list.
Cursor	`cur.close()`	Closes the cursor and releases resources.

Implementation Sample: Checking MySQL Driver API Level and Standard Transaction Processing

The following is sample code using the MySQL driver MySQLdb. After checking the API level, it performs transaction processing (including rollback when an exception occurs) assuming an access log management system.

Note: Execution requires the installation of the mysqlclient library and a running MySQL server.

import MySQLdb
import sys

# Database connection settings (Change according to your environment)
DB_CONFIG = {
    'host': 'localhost',
    'user': 'admin_user',
    'passwd': 'secure_password',
    'db': 'access_log_db',
    'charset': 'utf8mb4'
}

def check_api_compliance():
    """Check the driver's DB-API compliance level"""
    # The apilevel attribute indicates the DB-API version supported by the module
    # If '2.0' is displayed, it fully complies with the standard specification
    print(f"MySQLdb API Level: {MySQLdb.apilevel}")
    
    # Check thread safety level (1: Module cannot be shared, 2: Module can be shared, etc.)
    print(f"Thread Safety Level: {MySQLdb.threadsafety}")
    print("-" * 30)

def register_access_log(log_data):
    """
    Process to register access logs in the database
    Example implementation of connection management and transaction (commit/rollback)
    """
    conn = None
    try:
        # 1. Establish connection
        print("Connecting to database...")
        # Actual connection process (Commented out to prevent errors if no demo environment exists)
        # conn = MySQLdb.connect(**DB_CONFIG)
        
        # *Dummy object for explanation (In reality, use the result of connect above)
        class MockConnection:
            def cursor(self): return MockCursor()
            def commit(self): print(">> Commit successful.")
            def rollback(self): print(">> Rollback executed.")
            def close(self): print(">> Connection closed.")
        
        class MockCursor:
            def execute(self, sql, params): print(f"Executing SQL: {sql} with {params}")
            def close(self): print("Cursor closed.")

        conn = MockConnection() # Dummy connection

        # 2. Create cursor
        cur = conn.cursor()

        try:
            # 3. Execute SQL
            sql = "INSERT INTO server_logs (user_id, endpoint, status) VALUES (%s, %s, %s)"
            
            # Assuming multiple records processing
            for log in log_data:
                cur.execute(sql, (log['id'], log['path'], log['code']))
            
            # Test for intentional error (Uncomment if needed)
            # raise Exception("Simulation of database error")

            # 4. Commit (Confirm)
            conn.commit()
            print("All logs registered successfully.")

        except Exception as e:
            # 5. Rollback (Cancel)
            print(f"Error occurred: {e}")
            conn.rollback()
        
        finally:
            # 6. Close cursor
            cur.close()

    except MySQLdb.Error as e:
        print(f"Database connection failed: {e}")
    
    finally:
        # 7. Close connection (Written in finally block to ensure execution)
        if conn:
            conn.close()

def main():
    # Check DB-API level
    check_api_compliance()

    # Test data
    new_logs = [
        {'id': 101, 'path': '/api/login', 'code': 200},
        {'id': 102, 'path': '/api/dashboard', 'code': 200},
        {'id': 105, 'path': '/admin/settings', 'code': 403}
    ]

    # Execute main process
    register_access_log(new_logs)

if __name__ == "__main__":
    main()

Key Points

Pay attention to the try...except...finally block in the code. In database operations, it is important to structure your code to call rollback() if an error occurs midway to prevent data inconsistency.

Also, strictly call close() in the finally block to release connection resources to the server.

よかったらシェアしてね！