本文主要是介绍Python操作数据库的ORM框架SQLAlchemy快速入门教程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
连接内存版SQLIte
from sqlalchemy import create_engineengine = create_engine('sqlite:///:memory:')
print(engine)连接文件版SQLite
from sqlalchemy import create_engineengine = create_engine('sqlite:///sqlite3.db')
print(engine)
连接MySQL数据库
from sqlalchemy import create_engineengine = create_engine('mysql+pymysql://root:zhangdapeng520@127.0.0.1:3306/fastzdp_sqlalchemy?charset=utf8')
print(engine)
根据模型自动创建表
import enum
from datetime import datetime
from decimal import Decimalimport sqlalchemy
from sqlalchemy import create_engine, DateTime, func, String
from sqlalchemy.orm import Mapped, DeclarativeBase, mapped_columnengine = create_engine('mysql+pymysql://root:zhangdapeng520@127.0.0.1:3306/fastzdp_sqlalchemy?charset=utf8')class BaseModel(DeclarativeBase):"""基础模型"""id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), comment="创建时间")update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), onupdate=func.now(),comment="更新时间")class GenderEnum(enum.Enum):MALE = "男"FEMALE = "女"class Employee(BaseModel):"""员工模型，对应员工表"""__tablename__ = 'employee'name: Mapped[str] = mapped_column(String(36), index=True, nullable=False, comment="姓名")age: Mapped[int] = mapped_column(comment="年龄")salary: Mapped[Decimal] = mapped_column(sqlalchemy.DECIMAL, nullable=False, comment="薪资")bonus: Mapped[float] = mapped_column(sqlalchemy.FLOAT, default=0, comment="奖金")is_leave: Mapped[bool] = mapped_column(sqlalchemy.Boolean, default=False, comment="是否离职")gender: Mapped[GenderEnum] = mapped_column(sqlalchemy.String(6), default=GenderEnum.MALE, comment="性别")if __name__ == '__main__':BaseModel.metadata.drop_all(engine)BaseModel.metadata.create_all(engine)
通过session新增数据
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:employees = [Employee(name="张三1", age=23, salary=Decimal(30000), gender=GenderEnum.MALE.value),Employee(name="张三2", age=23, salary=Decimal(30000), gender=GenderEnum.MALE.value),Employee(name="张三3", age=23, salary=Decimal(30000), gender=GenderEnum.MALE.value),]session.add_all(employees)
根据ID查询
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)
这篇关于Python操作数据库的ORM框架SQLAlchemy快速入门教程的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
