- 连接数据库案例:
- import mysql.connector
- cnx = mysql.connector.connect(user=‘scott‘, password=‘tiger‘,
- host=‘127.0.0.1‘,
- database=‘employees‘)
- cnx.close()
- 或者
- from mysql.connector import (connection)
- cnx = connection.MySQLConnection(user=‘scott‘, password=‘tiger‘,
- host=‘127.0.0.1‘,
- database=‘employees‘)
- 异常处理
- import mysql.connector
- from mysql.connector import errorcode
- try:
- cnx = mysql.connector.connect(user=‘scott‘,
- database=‘testt‘)
- except mysql.connector.Error as err:
- if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
- print("Something is wrong with your user name or password")
- elif err.errno == errorcode.ER_BAD_DB_ERROR:
- print("Database does not exist")
- else:
- print(err)
- else:
- cnx.close()
- 很多参数时
- import mysql.connector
- config = {
- ‘user‘: ‘scott‘,
- ‘password‘: ‘tiger‘,
- ‘host‘: ‘127.0.0.1‘,
- ‘database‘: ‘employees‘,
- ‘raise_on_warnings‘: True,
- }
- cnx = mysql.connector.connect(**config)
- cnx.close()
- 使用拓展的连接方式
- import mysql.connector
- cnx = mysql.connector.connect(user=‘scott‘, password=‘tiger‘,
- host=‘127.0.0.1‘,
- database=‘employees‘,
- use_pure=False)
- cnx.close()
- 或者
- import mysql.connector
- config = {
- ‘user‘: ‘scott‘,
- ‘password‘: ‘tiger‘,
- ‘host‘: ‘127.0.0.1‘,
- ‘database‘: ‘employees‘,
- ‘raise_on_warnings‘: True,
- ‘use_pure‘: False,
- }
- cnx = mysql.connector.connect(**config)
- cnx.close()
- DDL语句的使用例子
- 创建数据库和表
- from __future__ import print_function
- import mysql.connector
- from mysql.connector import errorcode
- DB_NAME = ‘employees‘
- TABLES = {}
- TABLES[‘employees‘] = (
- "CREATE TABLE `employees` ("
- " `emp_no` int(11) NOT NULL AUTO_INCREMENT,"
- " `birth_date` date NOT NULL,"
- " `first_name` varchar(14) NOT NULL,"
- " `last_name` varchar(16) NOT NULL,"
- " `gender` enum(‘M‘,‘F‘) NOT NULL,"
- " `hire_date` date NOT NULL,"
- " PRIMARY KEY (`emp_no`)"
- ") ENGINE=InnoDB")
- TABLES[‘departments‘] = (
- "CREATE TABLE `departments` ("
- " `dept_no` char(4) NOT NULL,"
- " `dept_name` varchar(40) NOT NULL,"
- " PRIMARY KEY (`dept_no`), UNIQUE KEY `dept_name` (`dept_name`)"
- ") ENGINE=InnoDB")
- TABLES[‘salaries‘] = (
- "CREATE TABLE `salaries` ("
- " `emp_no` int(11) NOT NULL,"
- " `salary` int(11) NOT NULL,"
- " `from_date` date NOT NULL,"
- " `to_date` date NOT NULL,"
- " PRIMARY KEY (`emp_no`,`from_date`), KEY `emp_no` (`emp_no`),"
- " CONSTRAINT `salaries_ibfk_1` FOREIGN KEY (`emp_no`) "
- " REFERENCES `employees` (`emp_no`) ON DELETE CASCADE"
- ") ENGINE=InnoDB")
- TABLES[‘dept_emp‘] = (
- "CREATE TABLE `dept_emp` ("
- " `emp_no` int(11) NOT NULL,"
- " `dept_no` char(4) NOT NULL,"
- " `from_date` date NOT NULL,"
- " `to_date` date NOT NULL,"
- " PRIMARY KEY (`emp_no`,`dept_no`), KEY `emp_no` (`emp_no`),"
- " KEY `dept_no` (`dept_no`),"
- " CONSTRAINT `dept_emp_ibfk_1` FOREIGN KEY (`emp_no`) "
- " REFERENCES `employees` (`emp_no`) ON DELETE CASCADE,"
- " CONSTRAINT `dept_emp_ibfk_2` FOREIGN KEY (`dept_no`) "
- " REFERENCES `departments` (`dept_no`) ON DELETE CASCADE"
- ") ENGINE=InnoDB")
- TABLES[‘dept_manager‘] = (
- " CREATE TABLE `dept_manager` ("
- " `dept_no` char(4) NOT NULL,"
- " `emp_no` int(11) NOT NULL,"
- " `from_date` date NOT NULL,"
- " `to_date` date NOT NULL,"
- " PRIMARY KEY (`emp_no`,`dept_no`),"
- " KEY `emp_no` (`emp_no`),"
- " KEY `dept_no` (`dept_no`),"
- " CONSTRAINT `dept_manager_ibfk_1` FOREIGN KEY (`emp_no`) "
- " REFERENCES `employees` (`emp_no`) ON DELETE CASCADE,"
- " CONSTRAINT `dept_manager_ibfk_2` FOREIGN KEY (`dept_no`) "
- " REFERENCES `departments` (`dept_no`) ON DELETE CASCADE"
- ") ENGINE=InnoDB")
- TABLES[‘titles‘] = (
- "CREATE TABLE `titles` ("
- " `emp_no` int(11) NOT NULL,"
- " `title` varchar(50) NOT NULL,"
- " `from_date` date NOT NULL,"
- " `to_date` date DEFAULT NULL,"
- " PRIMARY KEY (`emp_no`,`title`,`from_date`), KEY `emp_no` (`emp_no`),"
- " CONSTRAINT `titles_ibfk_1` FOREIGN KEY (`emp_no`)"
- " REFERENCES `employees` (`emp_no`) ON DELETE CASCADE"
- ") ENGINE=InnoDB")
- cnx = mysql.connector.connect(user=‘scott‘)
- cursor = cnx.cursor()
- def create_database(cursor):
- try:
- cursor.execute(
- "CREATE DATABASE {} DEFAULT CHARACTER SET ‘utf8‘".format(DB_NAME))
- except mysql.connector.Error as err:
- print("Failed creating database: {}".format(err))
- exit(1)
- try:
- cnx.database = DB_NAME
- except mysql.connector.Error as err:
- if err.errno == errorcode.ER_BAD_DB_ERROR:
- create_database(cursor)
- cnx.database = DB_NAME
- else:
- print(err)
- exit(1)
- for name, ddl in TABLES.iteritems():
- try:
- print("Creating table {}: ".format(name), end=‘‘)
- cursor.execute(ddl)
- except mysql.connector.Error as err:
- if err.errno == errorcode.ER_TABLE_EXISTS_ERROR:
- print("already exists.")
- else:
- print(err.msg)
- else:
- print("OK")
- cursor.close()
- cnx.close()
- 插入数据
- from __future__ import print_function
- from datetime import date, datetime, timedelta
- import mysql.connector
- cnx = mysql.connector.connect(user=‘scott‘, database=‘employees‘)
- cursor = cnx.cursor()
- tomorrow = datetime.now().date() + timedelta(days=1)
- add_employee = ("INSERT INTO employees "
- "(first_name, last_name, hire_date, gender, birth_date) "
- "VALUES (%s, %s, %s, %s, %s)")
- add_salary = ("INSERT INTO salaries "
- "(emp_no, salary, from_date, to_date) "
- "VALUES (%(emp_no)s, %(salary)s, %(from_date)s, %(to_date)s)")
- data_employee = (‘Geert‘, ‘Vanderkelen‘, tomorrow, ‘M‘, date(1977, 6, 14))
- # Insert new employee
- cursor.execute(add_employee, data_employee)
- emp_no = cursor.lastrowid
- # Insert salary information
- data_salary = {
- ‘emp_no‘: emp_no,
- ‘salary‘: 50000,
- ‘from_date‘: tomorrow,
- ‘to_date‘: date(9999, 1, 1),
- }
- cursor.execute(add_salary, data_salary)
- # Make sure data is committed to the database
- cnx.commit()
- cursor.close()
- cnx.close()
- 查询数据
- import datetime
- import mysql.connector
- cnx = mysql.connector.connect(user=‘scott‘, database=‘employees‘)
- cursor = cnx.cursor()
- query = ("SELECT first_name, last_name, hire_date FROM employees "
- "WHERE hire_date BETWEEN %s AND %s")
- hire_start = datetime.date(1999, 1, 1)
- hire_end = datetime.date(1999, 12, 31)
- cursor.execute(query, (hire_start, hire_end))
- for (first_name, last_name, hire_date) in cursor:
- print("{}, {} was hired on {:%d %b %Y}".format(
- last_name, first_name, hire_date))
- cursor.close()
- cnx.close()
- 查询案例
- import mysql.connector
- from mysql.connector import errorcode
- config = {
- ‘user‘:‘test‘,
- ‘password‘:‘test‘,
- ‘host‘:‘127.0.0.1‘,
- ‘port‘:‘3306‘,
- ‘database‘:‘test‘,
- }
- try:
- con = mysql.connector.connect(**config)
- except mysql.connector.Error as err:
- if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
- print "Sometine is wrong with your user name or password"
- elif err.errno == errorcode.ER_BAD_DB_ERROR:
- print "Database does not exist"
- else:
- print(err)
- cursor = con.cursor()
- cursor.execute("select * from students")
- for (no, name, age, sex) in cursor:
- print "{} {} {} {}".format(no, name, age, sex)
时间: 2024-11-09 08:26:57