赞
踩
SQLAlchemy是用Python编程语言开发的一个开源项目。它提供了SQL工具包和ORM(对象关系映射)工具,使用MIT许可证发行。
SQLAlchemy采用简单的Python语言,提供高效和高性能的数据库访问,实现了完整的企业级持久模型。它的理念是SQL数据库的量级和性能比对象集合重要,而对象集合的抽象又重要于表和行。
- pip install pymysql
- pip install sqlalchemy
- 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()
- 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)
-
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。