with Session(engine)as session:session.begin()try:session.add(Employee(name="张三", age=23, salary=Decimal(30000),gender=GenderEnum.MALE.value))except:session.rollback()session.commit()
通过sessionmaker添加数据
with sessionmaker(engine).begin()as session:session.add(Employee(name="李四", age=23, salary=Decimal(30000), gender=GenderEnum.MALE.value))
with sessionmaker(engine).begin()as session:employee = session.get(Employee,1)print(employee.name)
查询所有的数据
with sessionmaker(engine).begin()as session:query = select(Employee)data = session.scalars(query).all()print(data)for employee in data:print(employee.name, employee.age)
查询指定字段
with sessionmaker(engine).begin()as session:query = select(Employee.id, Employee.name, Employee.age)data = session.execute(query).all()print(data)for employee in data:# rowprint(employee.name, employee.age)
执行原生SQL语句进行查询
with sessionmaker(engine).begin()as session:query = sqlalchemy.text("select id,name,age from employee")data = session.execute(query).all()print(data)for employee in data:# rowprint(employee.name, employee.age)
根据ID修改数据
with sessionmaker(engine).begin()as session:employee = session.get(Employee,1)employee.name ="张三333"
执行update方法
with sessionmaker(engine).begin()as session:query = sqlalchemy.update(Employee).where(Employee.id==1).values(name="张三", age=33)session.execute(query)
根据ID删除数据
with sessionmaker(engine).begin()as session:employee = session.get(Employee,1)session.delete(employee)
执行delete方法
with sessionmaker(engine).begin()as session:query = sqlalchemy.delete(Employee).where(Employee.id==2)session.execute(query)
执行is null查询
with sessionmaker(engine).begin()as session:query = select(Employee).where(Employee.salary.is_(None))# is nullemployees = session.execute(query).scalars()print(employees)
执行is not null查询
with sessionmaker(engine).begin()as session:query = select(Employee).where(Employee.salary.isnot(None))# is not nullemployees = session.execute(query).scalars()print(employees)for employee in employees:print(employee.name, employee.age, employee.salary, employee.bonus, employee.is_leave)
执行like模糊查询
with sessionmaker(engine).begin()as session:query = select(Employee).where(Employee.name.like("%3"))# like 模糊查询employees = session.execute(query).scalars()print(employees)for employee in employees:print(employee.name, employee.age, employee.salary, employee.bonus, employee.is_leave)
执行in查询
with sessionmaker(engine).begin()as session:query = select(Employee).where(Employee.id.in_([3,5]))# in 查询employees = session.execute(query).scalars()print(employees)for employee in employees:print(employee.name, employee.age, employee.salary, employee.bonus, employee.is_leave)
执行or查询
with sessionmaker(engine).begin()as session:query = select(Employee).where(sqlalchemy.or_(Employee.age <20, Employee.age >30))# or 查询employees = session.execute(query).scalars()print(employees)for employee in employees:print(employee.name, employee.age, employee.salary, employee.bonus, employee.is_leave)
求平均薪资
with sessionmaker(engine).begin()as session:query = select(func.avg(Employee.salary))avg = session.execute(query).first()print(avg)
统计表中的数据个数
with sessionmaker(engine).begin()as session:query = select(func.count(Employee.id))id_count = session.execute(query).first()print(id_count)
执行分页查询
with sessionmaker(engine).begin()as session:query = select(Employee).offset(2).limit(2)data = session.execute(query).scalars()for employee in data:print(employee.id, employee.name)
执行排序查询
with sessionmaker(engine).begin()as session:# query = select(Employee).order_by(Employee.age.desc()) # 降序query = select(Employee).order_by(Employee.age)# 升序data = session.execute(query).scalars()for employee in data:print(employee.id, employee.name, employee.age)
执行分组聚合查询
with sessionmaker(engine).begin()as session:query = select(Employee.gender, func.count(Employee.id)).group_by(Employee.gender)data = session.execute(query).all()for row in data:print(row.gender, row.count)