[Python] MySQL CRUD 샘플 스크립트

 

https://dev.mysql.com/doc/connector-python/en/

아주 간단한 MySQL 연동 샘플 입니다. mysql.connector 사용하며
데이터베이스는 mysql online 에서 제공하는 employee 데이터베이스를 이용 하였습니다.

#!/usr/bin/python #!-*- coding: utf-8 -*- import os import sys import time import logging import datetime import mysql.connector import logging.handlers from mysql.connector import errorcode from argparse import ArgumentParser from argparse import RawTextHelpFormatter from datetime import date, datetime, timedelta class mysql_api():  info = {   'user': 'fabric2',   'password': 'fabric2',   'host': '192.168.26.236',     'port': 3306,   'database': 'employees',   'raise_on_warnings': True,   'pool_name' : 'mypool',   'pool_size' : 2 ,                 'autocommit' : True,                 'compress': True   }  def __init__(self):   self.logger      = logging.getLogger("mysql")   self.logger.setLevel(logging.DEBUG)   formatter   = logging.Formatter('%(levelname)-6s %(asctime)s # %(message)s')   fileHandler = logging.FileHandler('mysql_stress.log')   fileHandler.setFormatter(formatter)   self.logger.addHandler(fileHandler)   self.mysql_conn = mysql.connector.connect(**self.info)   self.cursor=self.mysql_conn.cursor()  def mysql_connection(self, config):   try :                  if config.get('user')     != None : self.info['user']     =config['user']                  if config.get('password') != None : self.info['password'] =config['password']                  if config.get('host')     != None : self.info['host']     =config['host']                  if config.get('port')     != None : self.info['port']     =config['port']                  if config.get('database') != None : self.info['database'] =config['database']    self.mysql_conn = mysql.connector.connect(**self.info)    self.cursor=self.mysql_conn.cursor()   except Exception as e:    print e                def mysql_disconnect(self):   if self.mysql_conn != None:    self.mysql_conn.close()  def insert(self, query):   try:    self.cursor.execute(query)   except Exception as e:    self.logger.info(e)    print e  def update(self, query):   try:    self.cursor.execute(query)   except Exception as e:    self.logger.info(e)    print e  def delete(self, query):   try:    self.cursor.execute(query)   except Exception as e:    self.logger.info(e)    print e  def select(self, query):   try:    self.cursor.execute(query)    return self.cursor   except Exception as e:    self.logger.info(e)    print e  def select_print(self,result):   for i in result:    for j in i:     print(j),    print "" if __name__ == '__main__':         db=mysql_api()  tomorrow = datetime.now().date() + timedelta(days=1)  try:          db.mysql_connection({"database":"employees"})                 table_name="employees"      query = ("""\    INSERT INTO %s (emp_no, first_name, last_name, hire_date, gender, birth_date)    VALUES (%d,'%s','%s','%s','%s','%s')\    """ % (table_name,10,'길동','홍','2016-12-15','M','1975-01-19'))                 #db.insert(query)   query = ("""\    UPDATE %s SET first_name = '%s'                         WHERE gender = '%s'""" % (table_name,'KIM','M'))                 #db.update(query)   query = ("""\    DELETE FROM %s                         WHERE emp_no = %s""" % (table_name,1))                 #db.delete(query)   query = (""" SELECT * FROM %s WHERE emp_no > %s""" % (table_name,1))                 result=db.select(query)                 db.select_print(result)  except Exception as e:   print(e)