Python 3: mysql-connector で MySQL に接続する

Table of Contents

Python 3 で mysql.connector を使って MySQL に接続した方法を紹介します。

環境設定

Ubuntu と Amazon Linux で試しました。

Ubuntu

Ubuntu 16.04.1 LTS でやりました。

次のようにして必要なパッケージをインストールします。

sudo apt-get install python3-mysql.connector

1	sudo apt-get install python3-mysql.connector

Amazon Linux

Python 3 とその pip パッケージはインストールされているものとします。 Python のバージョンは 3.4 です。

MySQL のサイトからファイルをダウンロードしてインストールします。 Platform は Platform Independent を選びました。また、ダウンロードしたバージョンは 2.1.3 でした。私が (yum や pip ではインストールできませんでした、きっとやり方はあると思うのですが。)

ダウンロードしたファイルを (scp などを用いて) サーバにコピーします。私はコピーしたファイルを /usr/local/src に移動し、 rootユーザになって展開しました。

そして、rootユーザで次のコマンドを実行します。

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

接続確認

python3を実行して Python コンソールを開き、次のコマンドを実行すると接続確認ができます。

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

うまくいけば、 Uptime が表示されます。

コードで接続してみる

たとえば次のようなコードで、 SELECT を実行できます。 Config クラスがないと動きません。接続情報を直接書いて Config の import をなくせばOKです。 SQLインジェクションの可能性も残っています。 Python経験が浅いのでコードの質もきっと低いです。さーせん。

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