Connect to OceanBase Cloud by using PyMySQL

Click here to download the project code, which is a package named python-pymysql.zip.

Decompress the package to obtain a folder named python-pymysql. The directory structure is as follows:

python-pymysql
├── config.py
├── test_sql.py
├── main.py
└── requirements.txt

The files and directories are described as follows:

• config.py: stores database connection configurations.

• test_sql.py: stores SQL statements.

• main.py: serves as the entry to the application and provides a simple CLI for basic database operations such as creating tables, inserting data, querying data, and dropping tables.

• requirements.txt: lists the Python libraries required for the project and their version requirements.

  Note

  The requirements.txt file in the package obtained in this topic lists only the PyMySQL library and its version requirements. You can run the sudo pip install -r requirements.txt command to automatically install the required library.

Code in config.py

Code in the config.py file defines the database connection information.

You need to specify the IP address, port number, username, password, database name, and character set for connecting to OceanBase Cloud.

The sample code is as follows:

DB_CONFIG = {
    'host': '$host',
    'port': $port,
    'user': '$user_name',
    'password': '$password',
    'database': '$database_name',
    'charset': 'utf8mb4'
}

The parameters are described as follows:

• host: the access address of OceanBase Cloud. The value is sourced from the -h parameter in the connection string.

• port: the access port of OceanBase Cloud. The value is sourced from the -P parameter in the connection string.

• user: the account name. The value is sourced from the -u parameter in the connection string.

• password: the account password. The value is sourced from the -p parameter in the connection string.

• database: the name of the database to be accessed. The value is sourced from the -D parameter in the connection string.

• charset: the character set for connecting to the database.

  Notice

  You need to modify the parameter values in the sample code based on the actual environment and database settings.

Code in test_sql.py

Code in the test_sql.py file obtained in this topic defines SQL statements for database operations such as creating tables, inserting data, querying data, and dropping tables. These SQL statements can be executed after the application connects to the database through PyMySQL.

Perform the following steps to configure the test_sql.py file:

1. Define an SQL statement for creating a table.

   Define an SQL statement that creates the test_pymysql table with three fields: id, name, and age, where id is an auto-increment primary key.

   The sample code is as follows:

   CREATE_TABLE_SQL = '''
   CREATE TABLE test_pymysql (
   id INT(11) NOT NULL AUTO_INCREMENT,
   name VARCHAR(128) NOT NULL,
   age INT(11) NOT NULL,
   PRIMARY KEY (id)
   )
   '''
   

2. Define an SQL statement for inserting data.

   Define an SQL statement that inserts three rows into the test_pymysql table, with each row containing the name and age fields. The value of each field is passed in as a placeholder %s when the SQL statement is executed.

   The sample code is as follows:

   INSERT_DATA_SQL = '''
   INSERT INTO test_pymysql (name, age) VALUES 
   (%s, %s),
   (%s, %s),
   (%s, %s)
   '''
   

3. Define an SQL statement for querying data.

   Define an SQL statement that queries all data from the test_pymysql table.

   The sample code is as follows:

   SELECT_DATA_SQL = '''
   SELECT * FROM test_pymysql
   '''
   

4. Define an SQL statement for dropping a table.

   Define an SQL statement that drops the test_pymysql table.

   The sample code is as follows:

   DROP_TABLE_SQL = '''
   DROP TABLE test_pymysql
   '''
   

Code in main.py

Code in the main.py file obtained in this topic calls the pymysql module to connect to OceanBase Cloud in the MySQL compatible mode, calls the logging module to output logs, and implements database operations such as creating tables, inserting data, querying data, and dropping tables.

Perform the following steps to configure the main.py file:

1. Import the required modules.

   1. Import the logging module.
   2. Import the pymysql module.
   3. Import the config.py module, which defines the database connection information.
   4. Import the test_sql.py module, which defines SQL statements for database operations.

   The sample code is as follows:

   import logging
   import pymysql
   from config import DB_CONFIG
   from test_sql import CREATE_TABLE_SQL, INSERT_DATA_SQL, SELECT_DATA_SQL, DROP_TABLE_SQL
   

2. Set the level and format of logging, and generate an INFO log to indicate that script execution starts.

   The sample code is as follows:

   logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
   logging.info('Start executing the script')
   

3. Define a function for creating a table.

   Define a function named create_table(), and generate an INFO log to indicate that table creation starts. Use the with statement to manage the lifecycle of database connections and cursor objects. This ensures that database connections and cursor objects can be securely closed without memory leaks. Execute the SQL statement for creating a table, commit the transaction, and generate a success log or roll back the transaction and generate an error log.

   The sample code is as follows:

   def create_table():
       logging.info('Start creating the table')
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   cursor.execute(CREATE_TABLE_SQL)
                   conn.commit()
                   logging.info('Table creation successful')
               except Exception as e:
                   conn.rollback()
                   logging.error('Table creation failed, Reason: %s' % e)
   

4. Define a function for inserting data.

   Define a function named insert_data(), and generate an INFO log to indicate that data insertion starts. Use the with statement to manage the lifecycle of database connections and cursor objects. This ensures that database connections and cursor objects can be securely closed without memory leaks. Execute the SQL statement for inserting data, commit the transaction, and generate a success log or roll back the transaction and generate an error log.

   The sample code is as follows:

   def insert_data():
       logging.info('Start inserting data')
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   data = [('John', 20), ('Lucy', 25), ('Tom', 30)]
                   flattened_data = [d for item in data for d in item]
                   cursor.executemany(INSERT_DATA_SQL, [flattened_data])
                   conn.commit()
                   logging.info('Data insertion successful')
               except Exception as e:
                   conn.rollback()
                   logging.error('Data insertion failed, Reason: %s' % e)
   

5. Define a function for querying data.

   Define a function named select_data() to query data from a database. Use the with statement to manage the lifecycle of database connections and cursor objects. This ensures that database connections and cursor objects can be securely closed without memory leaks. Use the execute() method to execute the SQL statement defined by SELECT_DATA_SQL to query data. Use the fetchall() method to obtain the query result and output the result row by row by using the FOR loop.

   The sample code is as follows:

   def select_data():
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               cursor.execute(SELECT_DATA_SQL)
               result = cursor.fetchall()
               for row in result:
                   print(row)
   

6. Define a function for dropping a table.

   Define a function named drop_table() that uses pre-defined database connection information (DB_CONFIG) and the SQL statement for dropping tables (DROP_TABLE_SQL). The function drops tables and prints logs to show the operation result. If the table fails to be dropped, an error message is printed.

   The sample code is as follows:

   def drop_table():
       logging.info('Start dropping the table')
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   cursor.execute(DROP_TABLE_SQL)
                   conn.commit()
                   logging.info('Table dropped successfully')
               except Exception as e:
                   conn.rollback()
                   logging.error('Table drop failed, Reason: %s' % e)
   

7. Define the entry to the application, which is mainly used to perform database operations.

   First, determine whether the current module is directly run as the main program. If yes, perform the following operations:

   1. Call the create_table() function to create a table.
   2. Call the insert_data() function to insert data into a table.
   3. Call the select_data() function to query data from a table.
   4. Call the drop_table() function to drop a table.

   The sample code is as follows:

   if __name__ == '__main__':
       create_table()
       insert_data()
       select_data()
       drop_table()
   

8. Generate an INFO log to indicate that script execution is complete.

   The sample code is as follows:

   logging.info('Script execution completed')
   

Complete code

# Create a database connection.
DB_CONFIG = {
    'host': '$host',
    'port': $port,
    'user': '$user_name',
    'password': '$password',
    'database': '$database_name',
    'charset': 'utf8mb4'
}

# Create a table.
CREATE_TABLE_SQL = '''
CREATE TABLE test_pymysql (
  id INT(11) NOT NULL AUTO_INCREMENT,
  name VARCHAR(128) NOT NULL,
  age INT(11) NOT NULL,
  PRIMARY KEY (id)
  )
'''

# Insert data.
INSERT_DATA_SQL = '''
INSERT INTO test_pymysql (name, age) VALUES 
(%s, %s),
(%s, %s),
(%s, %s)
'''

# Query data.
SELECT_DATA_SQL = '''
SELECT * FROM test_pymysql
'''

# Drop the table.
DROP_TABLE_SQL = '''
DROP TABLE test_pymysql
'''

import logging
import pymysql
from config import DB_CONFIG
from test_sql import CREATE_TABLE_SQL, INSERT_DATA_SQL, SELECT_DATA_SQL, DROP_TABLE_SQL

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logging.info('Start executing the script')

# Create a table.
def create_table():
    logging.info('Start creating the table')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                cursor.execute(CREATE_TABLE_SQL)
                conn.commit()
                logging.info('Table creation successful')
            except Exception as e:
                conn.rollback()
                logging.error('Table creation failed, Reason: %s' % e)

# Insert data.
def insert_data():
    logging.info('Start inserting data')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                data = [('John', 20), ('Lucy', 25), ('Tom', 30)]
                flattened_data = [d for item in data for d in item]
                cursor.executemany(INSERT_DATA_SQL, [flattened_data])
                conn.commit()
                logging.info('Data insertion successful')
            except Exception as e:
                conn.rollback()
                logging.error('Data insertion failed, Reason: %s' % e)

# Query data.
def select_data():
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            cursor.execute(SELECT_DATA_SQL)
            result = cursor.fetchall()
            for row in result:
                print(row)

# Drop the table.
def drop_table():
    logging.info('Start dropping the table')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                cursor.execute(DROP_TABLE_SQL)
                conn.commit()
                logging.info('Table dropped successfully')
            except Exception as e:
                conn.rollback()
                logging.error('Table drop failed, Reason: %s' % e)

if __name__ == '__main__':
    create_table()
    insert_data()
    select_data()
    drop_table()

logging.info('Script execution completed')