Python:sqlalchemy框架, sqlalchemy应用(创建表格记录员工信息,数据处理)

it2026-06-17  9

sqlalchemy框架

可以连接多种关系型数据库无需编写sql语句sqlalchemy采用ORM Object:对象Relationship:关系Mapping:映射 sqlalchemy将python的class与数据库的表映射,一个class对应一张表表中的记录与class的实例映射表的字段与class的类变量映射字段的数据类型与sqlalchemy的类映射

创建数据库

假设您正在为一个小型公司编写应用,需要用到数据库。数据库需要记录员工的信息,以及发工资的情况数据库中需要出现的字段:员工ID、姓名、性别、生日、联系方式、部门ID、部门名、工资日、基本工资、奖金、实发工资

数据库范式

所谓第一范式(1NF)是指在关系模型中,对域添加的一个规范要求,所有的域都应该是原子性的,即数据库表的每一列都是不可分割的原子数据项。在本例中,联系方式可以由家庭住址、电话号码、email等多项构成,需要继续拆分。

第二范式(2NF)是在第一范式(1NF)的基础上建立起来的,即满足第二范式(2NF)必须先满足第一范式(1NF)。第二范式(2NF)要求数据库表中的每个实例或记录必须可以被唯一地区分。选取一个能区分每个实体的属性或属性组,作为实体的唯一标识。在本例中,部门名、工资都不依赖于员工ID,所以应该拆成多张表:员工表、部门表、工资表

员工表:员工ID(主键)、姓名、性别、生日、email、部门ID

部门表:部门ID(主键)、部门名

工资表:id(主键)、工资日、员工ID、基本工资、奖金、实发工资

第三范式(3NF)是第二范式(2NF)的一个子集,即满足第三范式(3NF)必须满足第二范式(2NF)。简而言之,3NF要求所有的属性依赖于主键,而不能依赖其他非主属性。在本例中,实发工资是通过基本工资和奖金计算出来的,所以它不应该出现在数据库中。当需要时,临时计算。

MariaDB [nsd2005]> CREATE DATABASE tedu2005 DEFAULT CHARSET utf8mb4; [root@localhost day04]# pip3 install sqlalchemy[root@localhost zzg_pypkgs]# pip3 install sqlalchemy_pkgs/SQLAlchemy-1.2.14.tar.gz

sqlalchemy应用

https://yiyibooks.cn/

https://yiyibooks.cn/wizard/sqlalchemy_11/index.html

# dbconn.py from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建连接数据库的引擎 engine = create_engine( # mysql+pymysql://用户名:密码@服务器地址/数据库?参数字符串 'mysql+pymysql://root:mima@127.0.0.1/database?charset=utf8', encoding='utf8', # echo=True # 在屏幕上输出调试信息,生产环境不要打开 ) Base = declarative_base() # 生成映射类的基类 Session = sessionmaker(bind=engine) # 创建到数据库的会话连接类 # 声明映射类 class Department(Base): __tablename__ = 'departments' # 固定名称,声明该类与哪张表关联 id = Column(Integer, primary_key=True) dep_name = Column(String(20), unique=True) class Employee(Base): __tablename__ = 'employees' id = Column(Integer, primary_key=True) emp_name = Column(String(20)) email = Column(String(50)) birth_date = Column(Date) dep_id = Column(Integer, ForeignKey('departments.id')) class Salary(Base): __tablename__ = 'salary' id = Column(Integer, primary_key=True) date = Column(Date) emp_id = Column(Integer, ForeignKey('employees.id')) basic = Column(Integer) awards = Column(Integer) if __name__ == '__main__': # 如果库中无表则创建,有表则不执行 Base.metadata.create_all(engine) # sqlalchemy_crud.py from dbconn import Department, Employee, Session # 创建会话实例 session = Session() ################################################## # 通过会话实例对数据库进行增删改查 ################################################## # 新增一个部门 # hr = Department(id=1, dep_name='人事部') # session.add(hr) # 新增多个部门 # ops = Department(id=2, dep_name='运维部') # dev = Department(id=3, dep_name='开发部') # qs = Department(id=4, dep_name='测试部') # finance = Department(id=5, dep_name='财务部') # sales = Department(id=6, dep_name='销售部') # market = Department(id=7, dep_name='市场部') # session.add_all([ops, dev, qs, finance, sales, market]) ################################################## # 新增员工 # lb = Employee( # id=1, emp_name='刘备', email='lb@tedu.cn', # birth_date='1980-01-01', dep_id=1 # ) # gy = Employee( # id=2, emp_name='关羽', email='gy@tedu.cn', # birth_date='1981-01-01', dep_id=2 # ) # zf = Employee( # id=3, emp_name='张飞', email='zf@tedu.cn', # birth_date='1982-01-01', dep_id=2 # ) # zgl = Employee( # id=4, emp_name='诸葛亮', email='zgl@qq.com', # birth_date='1982-05-04', dep_id=1 # ) # zy = Employee( # id=5, emp_name='赵云', email='zy@qq.com', # birth_date='1982-06-01', dep_id=3 # ) # hz = Employee( # id=6, emp_name='黄忠', email='hz@163.com', # birth_date='1975-07-01', dep_id=4 # ) # wy = Employee( # id=7, emp_name='魏严', email='wy@163.com', # birth_date='1990-08-01', dep_id=5 # ) # session.add_all([lb, gy, zf, zgl, zy, hz, wy]) ################################################## # 查询:参数如果是类名,返回的是实例集 # qset1 = session.query(Department) # print(qset1) # 此时,qset1只是一条sql语句 # for dep in qset1: # 取数据时,才会真正的查询数据库 # print(dep.id, dep.dep_name) # 查询:参数是类变量,返回的是元组构成的查询集 # qset2 = session.query(Employee.emp_name, Employee.email) # for data in qset2: # print(data) # 排序 # qset3 = session.query(Department).order_by(Department.id) # for dep in qset3: # print(dep.id, dep.dep_name) # 降序排列 # qset4 = session.query(Department).order_by(-Department.id) # for dep in qset4: # print(dep.id, dep.dep_name) # 过滤 # qset4 = session.query(Department).filter(Department.id>=3, Department.id<=5) # for dep in qset4: # print(dep.id, dep.dep_name) # 模糊查询 # qset5 = session.query(Employee.emp_name, Employee.email)\ # .filter(Employee.email.like('%@tedu.cn')) # for data in qset5: # print(data) # in # qset6 = session.query(Department).filter(Department.id.in_([1, 3, 5])) # for dep in qset6: # print(dep.id, dep.dep_name) # not in # qset7 = session.query(Department).filter(~Department.id.in_([1, 3, 5])) # for dep in qset7: # print(dep.id, dep.dep_name) # not null # qset8 = session.query(Department).filter(Department.dep_name.isnot(None)) # for dep in qset8: # print(dep.id, dep.dep_name) # 多表查询: query参数先写Employees,join的时候参数为Department # qset9 = session.query(Employee.emp_name, Department.dep_name).join(Department) # for data in qset9: # print(data) ################################################## # 取数据时,除了使用for循环,还可以使用all和first方法 # all取出所有记录,放到列表中;first只取出第一个结果 # qset9 = session.query(Employee.emp_name, Department.dep_name).join(Department) # print(qset9.all()) # print(qset9.first()) ################################################## # 将人事部改为人力资源部 # qset10 = session.query(Department).filter(Department.dep_name=='人事部') # hr = qset10.first() # hr.dep_name = '人力资源部' ################################################## # 删除市场部 qset11 = session.query(Department).filter(Department.dep_name=='市场部') market = qset11.first() session.delete(market) ################################################## # 确认 session.commit() # 关闭会话 session.close()
最新回复(0)