一、sqlalchemy简介
SQLAlchemy是用Python编程语言开发的一个开源项目。它提供了SQL工具包和ORM（对象关系映射）工具，使用MIT许可证发行。
SQLAlchemy采用简单的Python语言，提供高效和高性能的数据库访问，实现了完整的企业级持久模型。它的理念是SQL数据库的量级和性能比对象集合重要，而对象集合的抽象又重要于表和行。
二、sqlalchemy的使用
1、安装
python">pip install pymysql
pip install sqlalchemy2、CRUD（Create Read Update Delete，增查改删）
python">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 = nameself.gender = genderself.birthday = birthdayself.phone = phoneself.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 Falseelse:return Truefinally: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 Falseelse:return Truefinally: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.nameemp_update.gender = emp.genderemp_update.birthday = emp.birthdayemp_update.phone = emp.phoneemp_update.address = emp.addresssession.commit()except Exception as err:print("修改操作失败，%s" % err)session.rollback()return Falseelse:return Truefinally:session.close()#7.4 查询所有
def findAllEmployee():try:res = session.query(Employee).all()except Exception as err:print("查询失败，%s" % err)else:return resfinally:session.close()3、使用unittest进行测试
python">from src.com import MysqlUtils
import unittestclass 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 = 17if(MysqlUtils.updateEmployee(emp)):print("修改成功")def test_findAll(self):res = MysqlUtils.findAllEmployee()for i in res:print(i.id, i.name)
