Sample program for connecting to an OceanBase database with PyMySQL

Download the python-pymysql sample project

PS C:\Windows\system32> python --version
Python 3.7.0
PS C:\Windows\system32> pip --version
pip 22.3.1 from d:\python\python37\lib\site-packages\pip (python 3.7)

obclient -h$host -P$port -u$user_name -p$password -D$database_name

obclient -hxxx.xxx.xxx.xxx -P3306 -utest_user001 -p****** -Dtest

Project code introduction

Click python-pymysql to download the project code. The code is in a compressed package named python-pymysql.zip.

After you decompress the package, a folder named python-pymysql is created. The folder structure is as follows:

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

File description:

• config.py: Stores the database connection information.

• test_sql.py: Stores the SQL statements.

• main.py: The main program entry point. It executes basic database operations, such as creating a table, inserting data, querying data, and deleting the table.

• requirements.txt: Stores the required Python packages and their version information for the project.

  Note

  This topic lists only the version requirement for the PyMySQL library. You can run the sudo pip install -r requirements.txt command to install it. This command automatically installs the required libraries.

config.py code introduction

In the config.py file that is provided in this topic

The code defines the following database connection information:

The database connection information includes the IP address, port, username, password, database, and character set.

Code:

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

Parameter description:

• host: The domain name for the OceanBase database connection.

• port: The port for the OceanBase database connection. The default port is 3306.

• user: The account used to connect to the tenant.

• password: The password for the account.

• database: The name of the database to connect to.

• charset: The character set used for the database connection.

test_sql.py code introduction

The test_sql.py file defines the SQL statements for database operations, such as creating a table, inserting data, querying data, and deleting the table. You can execute these statements with PyMySQL to perform these functions.

This file includes the following parts:

1. SQL statement for creating a table.

   This SQL statement creates the test_pymysql table. The table has three fields: id, name, and age. The id field is an auto-incrementing primary key.

   Code:

   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. SQL statement for inserting data.

   This SQL statement inserts data into the test_pymysql table. The statement inserts three records. Each record has two fields: name and age. The values for each field are passed using the %s placeholder when the SQL statement is executed.

   Code:

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

3. SQL statement for querying data.

   This SQL statement queries all data from the test_pymysql table.

   Code:

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

4. SQL statement for deleting a table.

   This SQL statement deletes the test_pymysql table.

   Code:

   DROP_TABLE_SQL = '''
   DROP TABLE test_pymysql
   '''

main.py code introduction

The main.py file calls the pymysql module to connect to the database and the logging module to output log information. It performs operations such as creating a table, inserting data, querying data, and deleting the table.

The main.py file includes the following parts:

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 the SQL statements for database operations.

   Code:

   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. This part of the code sets the level and format for logging. It also outputs an INFO-level log message to indicate that the script has started.

   Code:

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

3. Define the function to create a table.

   This code defines a function named create_table(). The function outputs an INFO-level log message to indicate that it is starting to create the table. It uses a with statement to manage the lifecycle of the database connection and cursor objects. This ensures that the connection and cursor objects are closed safely to prevent issues such as memory leaks. The function executes the SQL statement to create the table, commits the transaction, and then outputs a log message. If an error occurs, the function rolls back the transaction and outputs an error log message.

   Code:

   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 the function to insert data.

   This code defines a function named insert_data(). The function outputs an INFO-level log message to indicate that it is starting to insert data. It uses a with statement to manage the lifecycle of the database connection and cursor objects. This ensures that the connection and cursor objects are closed safely to prevent issues such as memory leaks. The function executes the SQL statement to insert data, commits the transaction, and then outputs a log message. If an error occurs, the function rolls back the transaction and outputs an error log message.

   Code:

   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 the function to query data.

   This code defines a function named select_data() to query data from the database. It uses a with statement to manage the lifecycle of the database connection and cursor objects. This ensures that the connection and cursor objects are closed safely to prevent issues such as memory leaks. The function uses the execute() method to run the SQL statement defined by SELECT_DATA_SQL. It uses the fetchall() method to retrieve the query results and a for loop to print each row.

   Code:

   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 the function to delete a table.

   This code defines a function named drop_table(). The function uses the predefined database connection information (DB_CONFIG) and the SQL statement for deleting the table (DROP_TABLE_SQL). The function executes the operation to delete the table and prints a log message to indicate success or failure. If the operation fails, the function prints an error message.

   Code:

   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 point of the program. It is used to execute the database operation functions.

   First, the code checks if the current module is running as the main program. If it is, the code performs the following operations:

   1. Calls the create_table() function to create the database table.

   2. Calls the insert_data() function to insert data into the table.

   3. Calls the select_data() function to query data from the table.

   4. Calls the drop_table() function to delete the database table.

   Code:

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

8. This code outputs an INFO-level log message to indicate that the script has finished executing.

   Code:

   logging.info('Script execution completed')

Complete code

config.py

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

test_sql.py

# Create 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
'''

# Delete table
DROP_TABLE_SQL = '''
DROP TABLE test_pymysql
'''

main.py

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 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)

# Delete 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')

References

• For more information about connecting to an OceanBase database, see Connection methods overview.