MariaDB Connector/Python¶

Creating the Schema¶

1. Create a test database if one does not exist with the CREATE DATABASE statement:

   CREATE DATABASE IF NOT EXISTS test;
   

2. Create tables in the test database for testing basic and advanced operations with CREATE TABLE statements:

   CREATE TABLE test.contacts (
      id INT PRIMARY KEY AUTO_INCREMENT,
      first_name VARCHAR(25),
      last_name VARCHAR(25),
      email VARCHAR(100)
   ) ENGINE=InnoDB;
   
   CREATE TABLE test.accounts (
      id INT PRIMARY KEY AUTO_INCREMENT,
      first_name VARCHAR(25),
      last_name VARCHAR(25),
      email VARCHAR(100),
      amount DECIMAL(15,2) CHECK (amount >= 0.0),
      UNIQUE (email)
   ) ENGINE=InnoDB;
   

Creating the User¶

1. Create a user account to test connectivity with the CREATE USER statement:

   CREATE USER 'connpy_test'@'localhost'
      IDENTIFIED BY 'passwd';
   

2. Ensure that the user account has privileges to access the tables with the GRANT statement:

   GRANT SELECT, INSERT, UPDATE, DELETE, DROP
      ON test.contacts
      TO 'connpy_test'@'localhost';
   
   GRANT SELECT, INSERT, UPDATE, DELETE, DROP
      ON test.accounts
      TO 'connpy_test'@'localhost';
   

Opening a Connection¶

The connect() function can be used to open a new connection.

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Connect to MariaDB Platform using the connect() function with the relevant attributes.

   import sys
   
   # Instantiate Connection
   try:
      conn = mariadb.connect(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306)
   except mariadb.Error as e:
      print(f"Error connecting to MariaDB Platform: {e}")
      sys.exit(1)
   

   • The sys package is also imported, so that sys.exit() can be called in the event that the connection fails.

   • The connection attributes are passed as keyword arguments to the connect() function.

   • The connect() function returns an instance of the connection class, which is assigned to the conn variable.

   Additional information on connect() arguments is available in the Reference Documentation.

Closing a Connection¶

MariaDB Connector/Python closes the connection as part of the class's destructor, which is executed when an instance of the class goes out of scope. This can happen in many cases, such as:

• When the program exits

• When the connection instance is defined in the local scope of a function, and the function returns

• When the connection instance is defined as an attribute of a custom class's instance, and the custom class's instance goes out of scope

Connections can also be explicitly closed using the close() method, which is helpful when the connection is no longer needed, but the variable is still in scope.

1. Explicitly close the connection by calling the close() method on the connection:

   # Close Connection
   conn.close()
   

Using a Cursor¶

MariaDB Connector/Python accesses the database through a cursor, which is obtained by calling the cursor() method on the connection. This object provides you with an interface for performing basic operations in this section.

1. Create a cursor by calling the cursor() method on the connection:

   # Instantiate Cursor
   cur = conn.cursor()
   

2. Perform basic operations by executing SQL statements with the cursor.

   The cursor provides two methods for executing SQL statements:

   Method	Description
   execute()	Executes a single SQL statement.
   executemany()	Executes the given SQL statement for each tuple in a list.

   Examples that use these methods are shown in the sections below.

Inserting Data¶

MariaDB Connector/Python adds data to the database by executing an INSERT statement. Whether you use the execute() or executemany() methods to do so depends on whether you want to add a single row or several rows together.

To add a single row to the table:

1. Define an add_contact() function that adds a new contact to the table:

   # Adds contact
   def add_contact(cur, first_name, last_name, email):
      """Adds the given contact to the contacts table"""
   
      cur.execute("INSERT INTO test.contacts(first_name, last_name, email) VALUES (?, ?, ?)",
         (first_name, last_name, email))
   

   • The execute() method is called on the cursor, which executes an INSERT statement to insert a new row into the table.

   • The query string is the first value specified to the execute() method.

   • The values for the new row are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

2. Call the add_contact() function:

   new_contact_fname = "J. R. R."
   new_contact_lname = "Tolkien"
   new_contact_email = "jrr.tolkien@example.edu"
   
   add_contact(cur,
      new_contact_fname,
      new_contact_lname,
      new_contact_email)
   

   • The function uses the cursor to execute an INSERT statement using the specified values.

To add multiple rows to the table:

1. Define an add_multiple_contacts() function that adds several new contacts to the table:

   # Add Multiple Rows
   def add_multiple_contacts(cur, data):
      """Adds multiple contacts to database from given data"""
   
      cur.executemany("INSERT INTO test.contacts(first_name, last_name, email) VALUES (?, ?, ?)",
         data)
   

   • The executemany() method is called on the cursor, which executes an INSERT statement to insert multiple new rows into the table.

   • The query string is the first value specified to the executemany() method.

   • The values for the new rows are specified to the executemany() method using a list of tuples.

   • The values in each tuple are substituted for the question marks (?) in the query string.

2. Call the add_multiple_contacts() function:

   # Initialize Data
   new_contacts = [
      ("Howard", "Lovecraft", "hp.lovecraft@example.net"),
      ("Flannery", "O'Connor", "flan.oconnor@example.com"),
      ("Walker", "Percy", "w.percy@example.edu")
   ]
   
   add_multiple_contacts(cur, new_contacts)
   

   • The function uses the cursor to execute an INSERT statement using the values in each tuple in the new_contacts list.

Selecting Data¶

1. Define a print_contacts() function that retrieves all contacts from the table and prints them to stdout:

   # Print List of Contacts
   def print_contacts(cur):
      """Retrieves the list of contacts from the database and prints to stdout"""
   
      # Initialize Variables
      contacts = []
   
      # Retrieve Contacts
      cur.execute("SELECT first_name, last_name, email FROM test.contacts")
   
      # Prepare Contacts
      for (first_name, last_name, email) in cur:
         contacts.append(f"{first_name} {last_name} <{email}>")
   
      # List Contacts
      print("\n".join(contacts))
   

   • The execute() method is called on the cursor, which executes a SELECT statement to select all rows from the table.

   • The query string is the only value specified to the execute() method in this case.

   • Once the query finishes executing, the cursor retrieves a list of the rows.

   • The cursor retrieves each row as a tuple.

   • A for loop is used to iterate the list of tuples. Other options for iterating results are to use the following methods: fetchall(), fetchmany(), and fetchone().

   • The values from each tuple are formatted and appended to a list.

   • After all results have been retrieved, the list is printed to stdout.

2. Call the print_contacts() function:

   print_contacts(cur)
   

   • The function uses the cursor to execute a SELECT statement, and the resulting values are printed.

Replacing Data¶

1. Define a replace_contact() function that replaces a contact with new data:

   # Replace Contact
   def replace_contact(cur, contact_id, first_name, last_name, email):
      """Replaces contact with the given `contact_id` with new values"""
   
      cur.execute("REPLACE INTO test.contacts VALUES (?, ?, ?, ?)",
         (contact_id, first_name, last_name, email))
   

   • The execute() method is called on the cursor, which executes a REPLACE statement to replace a row in the table.

   • The query string is the first value specified to the execute() method.

   • The new values for the row are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

   • If the given contact_id exists in the table, the row's existing values are replaced with the new values.

   • If the given contact_id does not exist in the table, a new row is inserted with the new values.

2. Call the replace_contact() function:

   replace_contact_id = 1
   replace_contact_fname = "John Ronald Reuel"
   replace_contact_lname = "Tolkien"
   replace_contact_email = "jrr.tolkien@example.edu"
   
   replace_contact(cur,
      replace_contact_id,
      replace_contact_fname,
      replace_contact_lname,
      replace_contact_email)
   

   • The function uses the cursor to execute a REPLACE statement using the specified values.

Updating Data¶

1. Define an update_contact_last_name() function that updates the last name of a contact that has the given email:

   # Update Last Name
   def update_contact_last_name(cur, email, last_name):
      """Updates last name of a contact in the table"""
   
      cur.execute("UPDATE test.contacts SET last_name=? WHERE email=?",
         (last_name, email))
   

   • The execute() method is called on the cursor, which executes an UPDATE statement to update a row in the table.

   • The query string is the first value specified to the execute() method.

   • The new values for the row and the values for the where clause are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

2. Call the update_contact_last_name() function:

   update_contact_email = "jrr.tolkien@example.edu"
   update_contact_lname = "Tolkien the Grey"
   
   update_contact_last_name(cur,
      update_contact_email,
      update_contact_lname)
   

   • The function uses the cursor to execute an UPDATE statement using the specified values.

Deleting Data¶

1. Define a remove_contact() function that deletes contacts with the given email:

   # Remove Contact from Database
   def remove_contact(cur, email):
      """Removes contacts from the database"""
   
      cur.execute("DELETE FROM test.contacts WHERE email = ?", (email, ))
   

   • The execute() method is called on the cursor, which executes a DELETE statement to delete a row in the table.

   • The query string is the first value specified to the execute() method.

   • The values for the where clause are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

2. Call the remove_contact() function:

   remove_contact_email = "jrr.tolkien@example.edu"
   
   remove_contact(cur,
      remove_contact_email)
   

   • The function uses the cursor to execute a DELETE statement using the specified values.

Truncating Data¶

1. Define a truncate_contacts() function that deletes all of our contacts:

   # Truncate Contacts
   def truncate_contacts(cur):
      """Removes all data from contacts table"""
   
      cur.execute("TRUNCATE test.contacts")
   

   • The execute() method is called on the cursor, which executes a TRUNCATE statement to truncate the table.

   • The query string is the only value specified to the execute() method in this case.

2. Call the truncate_contacts() function:

   truncate_contacts(cur)
   

   • The function uses the cursor to execute a TRUNCATE statement.

Disabling Auto-commit¶

To disable auto-commit and use explicit transactions, set the connection instance autocommit attribute to False.

# Disable Auto-commit
conn.autocommit = False

To then use explicit transactions, then MariaDB's standard transaction related statements can be executed with MariaDB Connector/Python using a cursor:

• BEGIN

• ROLLBACK

• COMMIT

Additionally, instances of the connection class provide the commit() and rollback() methods, which can be used instead.

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Define an add_account() function that adds a new account to the table:

   # Adds account
   def add_account(cur, first_name, last_name, email, amount):
      """Adds the given account to the accounts table"""
   
      cur.execute("INSERT INTO test.accounts(first_name, last_name, email, amount) VALUES (?, ?, ?, ?)",
         (first_name, last_name, email, amount))
   

   • The execute() method is called on the cursor, which executes an INSERT statement to insert a new row into the table.

   • The query string is the first value specified to the execute() method.

   • The values for the new row are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

3. Define an update_account_amount() function that updates the amount in an account associated with the given email:

   # Update Last Name
   def update_account_amount(cur, email, change):
      """Updates amount of an account in the table"""
   
      cur.execute("UPDATE test.accounts SET amount=(amount-?) WHERE email=?",
         (change, email))
   

   • The execute() method is called on the cursor, which executes an INSERT statement to update a row in the table.

   • The query string is the first value specified to the execute() method.

   • The new values for the row and the values for the where clause are specified to the execute() method using a tuple.

   • The values in the tuple are substituted for the question marks (?) in the query string.

4. Call these functions in a transaction:

   try:
      conn = mariadb.connect(
            user="connpy_test",
            password="passwd",
            host="localhost",
            port=3306)
   
      conn.autocommit = False
   
      cur = conn.cursor()
   
      new_account_fname = "John"
      new_account_lname = "Rockefeller"
      new_account_email = "john.rockefeller@example.com"
      new_account_amount = 418000000000.00
   
      add_account(cur,
         new_account_fname,
         new_account_lname,
         new_account_email,
         new_account_amount)
   
      new_account_change = 1000000.00
   
      update_account_amount(cur,
         new_account_email,
         new_account_change)
   
      conn.commit()
      conn.close()
   except Exception as e:
      print(f"Error commiting transaction: {e}")
   
      conn.rollback()
   

   • The code disables conn.autocommit, so explicit transactions are used.

   • These function calls should succeed without any issues, so the transaction will be committed.

5. Call these functions in a transaction again, but this time use values that will fail the table's CHECK constraint (which ensures that the final amount is greater than 0):

   # Module Import
   import mariadb
   
   try:
      conn = mariadb.connect(
            user="connpy_test",
            password="passwd",
            host="localhost",
            port=3306)
   
      conn.autocommit = False
   
      cur = conn.cursor()
   
      new_account_fname = "Charlie"
      new_account_lname = "Bucket"
      new_account_email = "charlie.bucket@example.com"
      new_account_amount = 0.50
   
      add_account(cur,
         new_account_fname,
         new_account_lname,
         new_account_email,
         new_account_amount)
   
      new_account_change = 1.00
   
      update_account_amount(cur,
         new_account_email,
         new_account_change)
   
      conn.commit()
      conn.close()
   except Exception as e:
      print(f"Error commiting transaction: {e}")
   
      conn.rollback()
   

   • The code disables conn.autocommit, so explicit transactions are used.

   • These function calls will fail due to the CHECK constraint. The account has an amount of 0.50, which the UPDATE statement attempts to reduce to -0.50, which is an invalid amount. The transaction fails and is rolled by with an error message:

     Error commiting transaction: CONSTRAINT `accounts.amount` failed for `test`.`accounts`
     

Creating Connection Pools¶

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Create a connection pool by instantiating the ConnectionPool class in your Python code.

   # Create Connection Pool
   def create_connection_pool():
      """Creates and returns a Connection Pool"""
   
      # Create Connection Pool
      pool = mariadb.ConnectionPool(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306,
         pool_name="web-app",
         pool_size=20
      )
   
      # Return Connection Pool
      return pool
   

The connection parameters, such as the user name and password can be reconfigured after creating the connection pool. The connection pool parameters (pool_name and pool_size) cannot be reconfigured after creating the pool. Renaming the pool or increasing its size requires creating a new ConnectionPool instance.

Additional information is available on connection and connection pool parameters in the Reference Documentation.

Getting Connections from the Pool¶

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Create a connection pool by instantiating the ConnectionPool class in your Python code.

   # Create Connection Pool
   def create_connection_pool():
      """Creates and returns a Connection Pool"""
   
      # Create Connection Pool
      pool = mariadb.ConnectionPool(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306,
         pool_name="web-app",
         pool_size=20
      )
   
      # Return Connection Pool
      return pool
   

   Additional information is available on connection and connection pool parameters in the Reference Documentation.

3. Get a connection from the pool using the get_connection() method:

   # Establish Pool Connection
   try:
       pconn = pool.get_connection()
   
   except mariadb.PoolError as e:
   
      # Report Error
      print(f"Error opening connection from pool: {e}")
   
      # Create New Connection as Alternate
      pconn = mariadb.connection(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306)
   

   • Connections are retrieved from the pool using the get_connection() method.

   • When all the connections in the pool are in use, the get_connection() method raises a PoolError exception.

   • If the connection pool fails to provide a connection, create a new instance of the connection class outside the connection pool. You can then optionally add the alternate connectionto the pool if you want to use it further. For more information, see Adding Existing Connection.

4. Create a cursor by calling the cursor() method on the connection:

   # Instantiate Cursor
   cur = pconn.cursor()
   

5. Perform basic operations by executing SQL statements with the cursor.

   The cursor provides two methods for executing SQL statements:

   Method	Description
   execute()	Executes a single SQL statement.
   executemany()	Executes the given SQL statement for each tuple in a list.

   Examples that use these methods are shown in the sections under Basic Operations.

Returning Connections to the Pool¶

1. Return the connection by calling the close() method on the connection:

   # Close Connection
   pconn.close()
   

   • Closing a connections from a connection pool does not actually close the connection. Instead, the connection returns the connection to the pool.

   • Once the connection is returned to the pool, other application threads can use it.

Adding Existing Connections to the Pool¶

In cases where you have created a separate connection, you can add it to a connection pool and let the pool maintain the connection separately, rather than closing it outright:

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Create a connection pool by instantiating the ConnectionPool class in your Python code.

   # Create Connection Pool
   def create_connection_pool():
      """Creates and returns a Connection Pool"""
   
      # Create Connection Pool
      pool = mariadb.ConnectionPool(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306,
         pool_name="web-app",
         pool_size=20
      )
   
      # Return Connection Pool
      return pool
   

   Additional information is available on connection and connection pool parameters in the Reference Documentation.

3. Connect to MariaDB Platform using the mariadb.connect() function with the relevant attributes.

   import sys
   
   # Instantiate Connection
   try:
      pconn = mariadb.connect(
         user="connpy_test",
         password="passwd",
         host="localhost",
         port=3306)
   except mariadb.Error as e:
      print(f"Error connecting to MariaDB Platform: {e}")
      sys.exit(1)
   

   • The sys package is also imported, so that sys.exit() can be called in the event that the connection fails.

   • The connection attributes are passed as keyword arguments to the connect() function.

   • The connect() function returns an instance of the connection class, which is assigned to the conn variable.

   Additional information is available on connection and connection pool parameters in the Reference Documentation.

4. Add the connection to the pool using the add_connection() method.

   try:
       pool.add_connection(pconn)
   
    except mariadb.PoolError as e:
   
       # Report Error
       print(f"Error adding connection to pool: {e}")
   

   • When the connection pool has reached the maximum pool size, the add_connection() method raises a PoolError exception.

Reconfiguring Connection Pools¶

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Create a connection pool by instantiating the ConnectionPool class in your Python code.

   # Create Connection Pool
   def create_connection_pool():
      """Creates and returns a Connection Pool"""
   
      # Create Connection Pool
      pool = mariadb.ConnectionPool(
         user="db_user",
         password="db_user_passwd",
         pool_name="web-app",
         pool_size=20
         host="localhost",
         port=3306
     )
   
      # Return Connection Pool
      return pool
   

   Additional information is available on connection and connection pool parameters in the Reference Documentation.

3. Reconfigure a connection pool using the set_config() method:

   # Reconfigure Connection Pool
   def reconfigure_connection_pool(pool):
      """Reconfigures and returns a Connection Pool"""
   
      # Configure Connection Pool
      pool.set_config(
         user="connpy_test",
         password="passwd",
         host="192.0.2.2",
         port=5009
      )
   
      # Return Connection Pool
      return pool
   

   • Connections must be added to the pool manually using the add_connection() method when using this method.

   • The Connection attributes (user, password, host, etc.) associated with the connection pool can be reconfigured.

   • The Connection Pool attributes (pool_name, pool_size, etc.) cannot be reconfigured.

   Additional information is available on connection and connection pool parameters in the Reference Documentation.

Retrieving Field Information for Query Results¶

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Define a select_contacts() function that retrieves all contacts from the table:

   # Print List of Contacts
   def select_contacts(cur):
      """Retrieves the list of contacts from the database"""
   
      # Retrieve Contacts
      cur.execute("SELECT first_name, last_name, email FROM test.contacts")
   

3. Define a get_field_info() function that prints the field information associated with the cursor:

   # Get field info from cursor
   def get_field_info(cur):
      """Retrieves the field info associated with a cursor"""
   
      field_info = mariadb.fieldinfo()
   
      field_info_text = []
   
      # Retrieve Column Information
      for column in cur.description:
         column_name = column[0]
         column_type = field_info.type(column)
         column_flags = field_info.flag(column)
   
         field_info_text.append(f"{column_name}: {column_type} {column_flags}")
   
      return field_info_text
   

4. Call these functions, then print the field information:

   try:
      conn = mariadb.connect(
            user="connpy_test",
            password="passwd",
            host="localhost",
            port=3306)
   
      cur = conn.cursor()
   
      select_contacts(cur)
   
      field_info_text = get_field_info(cur)
   
      print("Columns in query results:")
   
      print("\n".join(field_info_text))
   
      conn.close()
   
   except Exception as e:
      print(f"Error: {e}")
   

   The results should look like this:

   Columns in query results:
   first_name: VAR_STRING
   last_name: VAR_STRING
   email: VAR_STRING
   

Retrieving Field Information for All Tables¶

1. Import MariaDB Connector/Python:

   # Module Import
   import mariadb
   

2. Define a show_tables() function that executes the SHOW TABLES statement:

   # Get list of tables
   def show_tables(cur):
      """Retrieves the list of tables from the database"""
   
      table_list = []
   
      # Retrieve Contacts
      cur.execute("SHOW TABLES")
   
      for (table,) in cur.fetchall():
         table_list.append(table)
   
      return table_list
   

3. Define a get_field_info() function that prints the field information associated with the cursor:

   # Get field info from cursor
   def get_field_info(cur):
      """Retrieves the field info associated with a cursor"""
   
      field_info = mariadb.fieldinfo()
   
      field_info_text = []
   
      # Retrieve Column Information
      for column in cur.description:
         column_name = column[0]
         column_type = field_info.type(column)
         column_flags = field_info.flag(column)
   
         field_info_text.append(f"{column_name}: {column_type} {column_flags}")
   
      return field_info_text
   

4. Define a get_table_field_info() function that prints the field information associated with a table:

   # Get field info from cursor
   def get_table_field_info(cur, table):
      """Retrieves the field info associated with a table"""
   
      # Fetch Table Information
      cur.execute(f"SELECT * FROM {table} LIMIT 1")
   
      field_info_text = get_field_info(cur)
   
      return field_info_text
   

5. Call these functions, and then print the field information for each table:

   try:
      conn = mariadb.connect(
            user="connpy_test",
            password="passwd",
            host="localhost",
            port=3306,
            database="test")
   
      cur = conn.cursor()
   
      tables = show_tables(cur)
   
      for table in tables:
         field_info_text = get_table_field_info(cur, table)
   
         print(f"Columns in table {table}:")
         print("\n".join(field_info_text))
         print("\n")
   
      conn.close()
   
   except Exception as e:
      print(f"Error: {e}")
   

   The results should look like this:

   Columns in table accounts:
   id: LONG NOT_NULL | PRIMARY_KEY | AUTO_INCREMENT | NUMERIC
   first_name: VAR_STRING
   last_name: VAR_STRING
   email: VAR_STRING UNIQUE_KEY
   amount: NEWDECIMAL NUMERIC
   
   
   Columns in table contacts:
   id: LONG NOT_NULL | PRIMARY_KEY | AUTO_INCREMENT | NUMERIC
   first_name: VAR_STRING
   last_name: VAR_STRING
   email: VAR_STRING