使用 Python 操作 MySQL，这篇文章别错过~

1. 前言

2. 准备

• Python-MySql

• PyMysql

• SQLAlchemy

# 安装依赖
pip3 install pymysql

import pymysql

# 数据库连接
self.db = pymysql.connect(host='localhost',
                          port=3306,
                          user='root',
                          password='**',
                          database='xh')

# 获取游标
self.cursor = self.db.cursor()

# 插入一条数据
SQL_INSERT_A_ITEM = "INSERT INTO PEOPLE(name,age) VALUES('xag',23);"

def insert_a_item(self):
    """
    插入一条数据
    :return:
    """
    try:
        self.cursor.execute(SQL_INSERT_A_ITEM)
        self.db.commit()
    except Exception as e:
        print('插入数据失败')
        print(e)
        self.db.rollback()

# 插入多条数据SQL，name和age是变量，对应列表
SQL_INSERT_MANY_ITEMS = "INSERT INTO PEOPLE (name, age) VALUES(%s, %s)"

# 待插入的数据
self.datas = [("张三", 23), ("李四", 24), ("王五", 25)]

def insert_items(self):
    """
    插入多条记录
    :return:
    """
    try:
        self.cursor.executemany(SQL_INSERT_MANY_ITEMS, self.datas)
        self.db.commit()
    except Exception as e:
        print("插入数据异常")
        self.db.rollback()

# 查询所有记录
SQL_QUERY_ALL = "SELECT * FROM PEOPLE;"

def query(self):
    """查询数据"""
    # 查询所有数据
    self.cursor.execute(SQL_QUERY_ALL)

    # 元组数据
    rows = self.cursor.fetchall()

    # 打印结果
    for row in rows:
        id = row[0]
        name = row[1]
        age = row[2]
        print('id:', id, ',name:', name, 'age:', age)

# 按id查询
SQL_QUERY_WITH_CONDITION = "SELECT * FROM PEOPLE WHERE id={};"

# 查询id为5的记录
self.cursor.execute(SQL_QUERY_WITH_CONDITION.format(5))

# 更新(通过id去更新)
SQL_UPDATE = "UPDATE PEOPLE SET name='%s',age=%s WHERE id=%s"

def update(self):
    """
    更新数据
    :return:
    """
    sql_update = SQL_UPDATE % ("王五五", 30, 5)
    print(sql_update)

    try:
        self.cursor.execute(sql_update)
        self.db.commit()
    except Exception as e:
        self.db.rollback()
        print('更新数据异常')
        print(e)

# 删除（通过id去删除数据）
SQL_DELETE = "DELETE FROM PEOPLE WHERE id=%d"

def delete(self):
    """
    删除记录
    :return:
    """
    try:
        # 删除的完整sql
        sql_del = SQL_DELETE % (5)
        self.cursor.execute(sql_del)
        self.db.commit()
    except Exception as e:
        # 发生错误时回滚
        self.db.rollback()
        print(e)

def teardown(self):
    # 释放资源
    self.cursor.close()
    self.db.close()

# 安装依赖包
pip3 install sqlalchemy

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

# 基础类
Base = declarative_base()


# 自定义的表
class People(Base):
    # 表名
    __tablename__ = 'people'

    # 定义字段
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    age = Column(Integer)

    def __repr__(self):
        """
        便于打印结果
        :return:
        """
        return "<People(id:{},name:{},age:{})".format(self.id, self.name, self.age)

# 创建数据库的引擎实例对象
# 数据库名称：xh
engine = create_engine("mysql+pymysql://root:数据库密码@localhost:3306/xh",
                               encoding="utf-8",
                               echo=True)

# 创建表结构
# checkfirst：判断表是否存在，如果存在，就不重复创建
Base.metadata.create_all(engine, checkfirst=True)

# 实例化会话
self.session = sessionmaker(bind=engine)()

def add_item(self):
    """
    新增
    :return:
    """
    # 实例化一个对象
    people = People(name='xag', age=23)
    self.session.add(people)

    # 提交数据才会生效
    self.session.comit()

def add_items(self):
    """
    新增多条记录
    :return:
    """
    datas = [
        People(name='张三', age=20),
        People(name='李四', age=21),
        People(name='王五', age=22),
    ]
    self.session.add_all(datas)

    self.session.commit()

def query(self):
    """
    查询
    :return:
    """
    # 查询所有记录
    # result = self.session.query(People).all()

    # 查询name/age两个字段
    result = self.session.query(People.name, People.age).all()
    print(result)

# 条件查询
resp = self.session.query(People).filter_by(name='xag').first()
print(resp)

def update1(self, id):
    """
    更新数据1
    :return:
    """
    # 获取数据
    temp_people = self.session.query(People).filter_by(id=id).first()

    # 更新数据
    temp_people.name = "星安果"
    temp_people.age = 18
    # 提交修改
    self.session.commit()

def update2(self, id):
    """
    更新数据2
    :param id:
    :return:
    """
    # 使用update()方法直接更新字段值
    self.session.query(People).filter(People.id == id).update({People.name: "xag", People.age: 1})
    self.session.commit()

def del_by_id(self, id):
    """
    通过id删除一条记录
    :param id:
    :return:
    """
    del_count = self.session.query(People).filter(People.id == id).delete()
    print('删除数目:', del_count)
    self.session.commit()

5.最后