Python 3: Connect to MySQL with mysql-connector

Table of Contents

Here, I will introduce how to connect to MySQL using mysql.connector in Python 3.

Environment Setup

I have tested this on both Ubuntu and Amazon Linux.

Ubuntu

I performed this on Ubuntu 16.04.1 LTS.

To install the necessary packages, run the following command:

sudo apt-get install python3-mysql.connector

1	sudo apt-get install python3-mysql.connector

Amazon Linux

Assuming Python 3 and its pip package are already installed. The Python version is 3.4.

Download the file from MySQL’s site and install it. I chose “Platform Independent” as the platform, and the downloaded version was 2.1.3. (I couldn’t install it using yum or pip, but there should be a way.)

Copy the downloaded file to the server (you can use scp or similar methods). I moved the copied file to /usr/local/src and extracted it as the root user.

Then, execute the following command as the root user:

cd /usr/local/src/mysql-connector-python-2.1.3
python3 setup.py install

1 2	cd /usr/local/src/mysql-connector-python-2.1.3 python3 setup.py install

Connection Verification

You can verify the connection by running python3 to open the Python console and execute the following command:

import mysql.connector
config = {
  'user': 'app',
  'password': 'app',
  'host': 'rds.amazonaws.com',
  'database': 'development_db',
}
cnx = mysql.connector.connect(**config)
cur = cnx.cursor(buffered=True)
cur.execute("SHOW STATUS LIKE 'Uptime'")
print(cur.fetchone())
# => ('Uptime', '22073875')
cur.close()
# => True
cnx.close()

import mysql.connector

config = {

'user': 'app',

'password': 'app',

'host': 'rds.amazonaws.com',

'database': 'development_db',

}

cnx = mysql.connector.connect(**config)

cur = cnx.cursor(buffered=True)

cur.execute("SHOW STATUS LIKE 'Uptime'")

print(cur.fetchone())

# => ('Uptime', '22073875')

cur.close()

# => True

cnx.close()

If everything goes well, you should see ‘Uptime’ displayed.

Connecting via Code

For example, you can execute a SELECT statement with code like this. Note that it won’t work without the Config class. You can remove the import of Config and directly write connection information. Be cautious of SQL injection possibilities. My Python experience is limited, so the code quality might be low. Sorry about that.

import mysql.connector
from config.config import Config
from app.model.sample import Sample


class BaseMapper:
    @staticmethod
    def create_connection():
        return mysql.connector.connect(
            user=Config.get('database.user'),
            database=Config.get('database.name'),
            password=Config.get('database.pass'),
            host=Config.get('database.host'),
        )

    @staticmethod
    def execute(connection, sql, parameters = None):
        """

        :param connection:
        :param sql:
        :type sql: str
        :param parameters:
        :type parameters: tuple
        :return:
        """

        cursor = connection.cursor()
        cursor.execute((sql), parameters)

        results = []
        for item in cursor:
            results.append(item)
        cursor.close()
        return results


class SampleMapper(BaseMapper):
    def find_samples_by_name(self, connection, name):
        """

        :param name:
        :type name: int
        :return:
        """
        if connection is None:
            connection = self.create_connection()

        sql =
            "SELECT column_1, cplumn_2" 
            "  FROM samples" 
            " WHERE samples.name = " + name
        records = BaseMapper.execute(connection, sql, phrase_id)

        samples = []
        for record in records:
            sample = Sample()
            sample.column_1 = record[0]
            sample.column_2 = record[1]
            samples.append(sample)
        return samples

import mysql.connector

from config.config import Config

from app.model.sample import Sample

class BaseMapper:

@staticmethod

def create_connection():

return mysql.connector.connect(

user=Config.get('database.user'),

database=Config.get('database.name'),

password=Config.get('database.pass'),

host=Config.get('database.host'),

)

@staticmethod

def execute(connection, sql, parameters = None):

"""

:param connection:

:param sql:

:type sql: str

:param parameters:

:type parameters: tuple

:return:

"""

cursor = connection.cursor()

cursor.execute((sql), parameters)

results = []

for item in cursor:

results.append(item)

cursor.close()

return results

class SampleMapper(BaseMapper):

def find_samples_by_name(self, connection, name):

"""

:param name:

:type name: int

:return:

"""

if connection is None:

connection = self.create_connection()

sql =

"SELECT column_1, cplumn_2"

" FROM samples"

" WHERE samples.name = " + name

records = BaseMapper.execute(connection, sql, phrase_id)

samples = []

for record in records:

sample = Sample()

sample.column_1 = record[0]

sample.column_2 = record[1]

samples.append(sample)

return samples

The Life