MariaDB Python Application

Create a MariaDB Python Test Script on Rocky Linux 9.

In this blog, we are going to prepare a basic Linux server using a Python script to test connectivity to a MariaDB Database Server. The following steps are completed on a Rocky Linux 9 server but should work on most other versions of Linux, for example, CentOS 7/8/9.

Before following this blog post, please make sure you have an application server running with the MariaDB Connector installed, as detailed here.

To allow the application to connect to the MaxScale servers, a user must be created on the primary database server. We can do this via MaxScale, a GUI tool, or directly on the server. To create the user, log in using an existing user that has the correct permission to create an application test user.

The subnet ‘10.106.0.%’ will need to be modified. Avoid using ‘%’ as a default and use your application server IP address instead:

CREATE USER IF NOT EXISTS 'pythontest'@'10.106.0.%' IDENTIFIED BY 'aBcd123_';
GRANT SELECT, INSERT, UPDATE, CREATE, DROP ON *.* TO 'pythontest'@'10.106.0.%';

Once you have the connector and Python installed, we will write a Python script that will create the required database and a test table. Use whichever editor you prefer, for example, vi:

vi ~/mariadbCreate.py

Insert this code.

Note: the user and password must match the user you created in the earlier steps, and the IP addresses (listed on line 7) needs to be that of MaxScale or Primary Database server.

#!/usr/bin/env python3
import mariadb
import time
def pydemo():
    try:
        conn = mariadb.connect(
             host="10.106.0.5",
             port=3306,
             user="pythontest",
             password="aBcd123_")
        conn.auto_reconnect = True
        cursor = conn.cursor()
        cursor.execute("DROP DATABASE IF EXISTS pydemo")
        cursor.execute("CREATE DATABASE pydemo")
        cursor.execute("CREATE TABLE pydemo.test (id SERIAL PRIMARY KEY, uuid UUID NOT NULL,"
                       "created DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP) "
                       "ENGINE=INNODB DEFAULT CHARSET=utf8")
        conn.commit()
        for x in range(100):
           starttime = time.perf_counter()
           cursor.execute("INSERT INTO pydemo.test(uuid) VALUES (UUID())")
           conn.commit()
           endtime = time.perf_counter() - starttime
           print ('{:.6f}s for the transaction to complete'.format(endtime))
           time.sleep(1)
        cursor.execute("SELECT count(*) FROM pydemo.test")
        row= cursor.fetchone()
        cursor.close()
        print(*row,"Records created")
        conn.commit()
        if (conn):
            conn.close()
            print("The MariaDB connection is now closed")
    except mariadb.Error as e:
        print(f"MariaDB Python Demo ERROR: {e}")
pydemo()

Make the script executable:

chmod +x ~/mariadbCreate.py

and then execute it:

~/mariadbCreate.py

The script will loop for the number of seconds defined by default 100. When completed, the command line will show the number of records inserted. For each inserted line of data, the script will show the transaction time.

Now that we have the code running for the test application, you can learn how to obtain Continuous Availability within the database layer to avoid any database downtime.