Python + SQLAlchemy操作MySQL数据库（ORM）

一、sqlalchemy简介

SQLAlchemy是用Python编程语言开发的一个开源项目。它提供了SQL工具包和ORM（对象关系映射）工具，使用MIT许可证发行。

SQLAlchemy采用简单的Python语言，提供高效和高性能的数据库访问，实现了完整的企业级持久模型。它的理念是SQL数据库的量级和性能比对象集合重要，而对象集合的抽象又重要于表和行。

二、sqlalchemy的使用

1、安装

pip install pymysql
pip  install  sqlalchemy

2、CRUD（Create Read Update Delete，增查改删）

from sqlalchemy  import create_engine  #导入数据库引擎
from sqlalchemy.ext.declarative import declarative_base #导入orm基础类
from sqlalchemy import Column,Integer,String,DateTime #导入orm的列、数据类型
from sqlalchemy.orm import sessionmaker #导入session(会话)生成器
 
#1.创建一个orm的基础类
Base = declarative_base()
 
#2.创建数据库连接的引擎(数据库+数据库操作包://用户名:密码@数据库服务器地址:端口号/数据库名)
engine = create_engine('mysql+pymysql://root:#####@localhost:3306/dbms')
 
#3.将数据库连接引擎绑定到Base对象上
Base.metadata.create_all(engine)
 
#4.创建Session,绑定数据库引擎
Session = sessionmaker(bind=engine)
 
#5.创建Session对象(会话对象)
session = Session()
 
#6.声明ORM映射类,必须继承Base
class Employee(Base):
    __tablename__ = "employee"
    id = Column(Integer,primary_key=True)
    name = Column(String(30))
    gender = Column(String(4))
    birthday = Column(DateTime)
    phone = Column(String(11))
    address = Column(String(50))
 
    def __init__(self,name,gender,birthday,phone,address):
        self.name = name
        self.gender = gender
        self.birthday = birthday
        self.phone = phone
        self.address = address
 
#7.定义单表的CRUD函数
#7.1 增加员工信息
def insertEmployee(emp):
    try:
        session.add(emp)
        session.commit()
    except Exception as err:
        print("增加操作失败，%s" % err)
        session.rollback()
        return False
    else:
        return True
    finally:
        session.close()
 
#7.2 删除员工信息
def deleteEmployee(emp_id):
    try:
        emp_to_delete = session.query(Employee).filter_by(id=emp_id).first()
        if emp_to_delete:
            session.delete(emp_to_delete)
            session.commit()
    except Exception as err:
        print("删除操作失败，%s" % err)
        session.rollback()
        return False
    else:
        return True
    finally:
        session.close()
 
#7.3 修改员工信息
def updateEmployee(emp):
    try:
        emp_update = session.query(Employee).filter_by(id=emp.id).first()
        if emp_update:
            emp_update.name = emp.name
            emp_update.gender = emp.gender
            emp_update.birthday = emp.birthday
            emp_update.phone = emp.phone
            emp_update.address = emp.address
            session.commit()
    except Exception as err:
        print("修改操作失败，%s" % err)
        session.rollback()
        return False
    else:
        return True
    finally:
        session.close()
 
#7.4 查询所有
def findAllEmployee():
    try:
        res = session.query(Employee).all()
    except Exception as err:
        print("查询失败，%s" % err)
    else:
        return res
    finally:
        session.close()

3、使用unittest进行测试

from  src.com import MysqlUtils
import unittest
 
 
class TestORM(unittest.TestCase):
 
    def test_insert(self):
        emp = MysqlUtils.Employee('AA','男','2001-10-15','13578956589','宁夏银川')
        if(MysqlUtils.insertEmployee(emp)):
            print("增加成功")
        print("---------------------------------------")
 
    def test_delete(self):
        if(MysqlUtils.deleteEmployee(26)):
            print("删除成功")
 
    def test_update(self):
        emp = MysqlUtils.Employee('刘备','男','2024-11-11','18991758746','湖北宜昌');
        emp.id = 17
        if(MysqlUtils.updateEmployee(emp)):
            print("修改成功")
 
    def test_findAll(self):
        res = MysqlUtils.findAllEmployee()
        for i in res:
            print(i.id, i.name)