您现在的位置是：首页 >学无止境 >SQLAlchemy最新2.0数据库相关操作网站首页学无止境

SQLAlchemy最新2.0数据库相关操作

敲代码敲到头发茂密 2025-02-23 12:01:02
简介SQLAlchemy最新2.0数据库相关操作

SQLAlchemy最新2.0数据库操作

一、Session对象

session⽤于创建程序和数据库之间的会话，所有对象的载⼊和保存都需通过session对象。在Web项⽬中，⼀个请求共⽤⼀个session对象。

1、创建Session对象的两种⽅式

# 第⼀种，需要⾃⼰提交事务
with Session(bind=engine) as session:
	 session.begin()
	 try:
		 session.add(some_object)
		 session.add(some_other_object)
	 except:
		 session.rollback()
	 	raise
	 else:
	 	session.commit()
# 第⼆种， 不需要⾃⼰提交事务
with sessionmaker(bind=engine).begin() as session:
 	sess.execute()

2、新增模型对象操作

add：添加单个对象
add_all：批量添加对象

    with sessionmaker(engine).begin() as session:
        # 新增数据（一）
        emp1 = Employee(name="kobe", sal=10000, bonus=1000, gender=SexValue.MALE, entry_date=date(2019, 1, 1))
        emp2 = Employee(name="james", sal=20000, bonus=2000, gender=SexValue.MALE, entry_date=date(2018, 1, 1))
        emp3 = Employee(name="curry", sal=30000, bonus=3000, gender=SexValue.MALE, entry_date=date(2017, 1, 1))
        emp4 = Employee(name="green", sal=30000, bonus=3000, gender=SexValue.MALE, entry_date=date(2017, 1, 1))

        # session.add(emp1)
        session.add_all([emp3,emp4])

类SQL的方式

    with sessionmaker(engine).begin() as session:
        insert_stmt=insert(Employee).values(name="durent",sal=11000)
        session.execute(insert_stmt)

3、简单查询操作

3.1、根据主键查询返回一条数据

    with sessionmaker(engine).begin() as session:
        # todo get返回一条数据
        emp1=session.get(Employee,1)
        print(emp1)

kobe, 男, 10000.00, 2019-01-01,1000

3.2、查询整张表的数据

a、返回模型对象
    with sessionmaker(engine).begin() as session:
        # todo get返回所有数据的所有字段,scalars返回的是模型对象
        stmt=select(Employee)
        result=session.scalars(stmt).all()
        print(result)
        for emp in result:
            print(emp)

2025-02-06 18:26:59,823 INFO sqlalchemy.engine.Engine [generated in 0.00011s] {}
[<__main__.Employee object at 0x00000232BD672AF0>, <__main__.Employee object at 0x00000232BD6851F0>, <__main__.Employee object at 0x00000232BD685250>, <__main__.Employee object at 0x00000232BD685280>, <__main__.Employee object at 0x00000232BD6852B0>]
kobe, 男, 10000.00, 2019-01-01,1000
james, 男, 20000.00, 2018-01-01,2000
curry, 男, 30000.00, 2017-01-01,3000
green, 男, 30000.00, 2017-01-01,3000
durent, 男, 11000.00, 2025-02-06,0
2025-02-06 18:26:59,824 INFO sqlalchemy.engine.Engine COMMIT
b、返回row对象，⼀般⽤于指定返回的字段
    with sessionmaker(engine).begin() as session:
        # todo 返回指定字段的数据，execute返回的是row对象,row对象相当于字典，key就是属性名名字，value是字段的值
        stmt=select(Employee.name,Employee.sal,Employee.gender)
        result=session.execute(stmt).all()
        for emp in result:
            print(type(emp))
            print(emp.sal,emp.name,emp.gender.value)

<class 'sqlalchemy.engine.row.Row'>
10000.00 kobe 男
<class 'sqlalchemy.engine.row.Row'>
20000.00 james 男
<class 'sqlalchemy.engine.row.Row'>
30000.00 curry 男
<class 'sqlalchemy.engine.row.Row'>
30000.00 green 男
<class 'sqlalchemy.engine.row.Row'>
11000.00 durent 男
2025-02-06 18:27:55,062 INFO sqlalchemy.engine.Engine COMMIT

进程已结束，退出代码为 0
c、执⾏原⽣的SQL，并返回row对象
    with sessionmaker(engine).begin() as session: 
        # todo 采用原生sql来查询
        # todo 返回指定字段的数据，execute返回的是row对象,row对象相当于字典，key就是属性名名字，value是字段的值
        sql = text("select id,emp_name,sal,gender from t_emp")
        result=session.execute(sql).all()
        for obj in result:
            print(type(obj))
            print(obj.id,obj.emp_name,obj.sal,obj.gender)
d、执⾏原⽣的sql，并返回模型对象
    with sessionmaker(engine).begin() as session:
        # todo 采用原生sql来查询,并返回模型对象
        # todo 返回指定字段的数据，execute返回的是row对象,row对象相当于字典，key就是属性名名字，value是字段的值
        sql_text = text('select id, sal, gender from t_emp')
        # todo 映射转化为模型类对象
        new_sql = sql_text.columns(Employee.id, Employee.name, Employee.sal,
                                   Employee.gender)
        stmt = select(Employee).from_statement(new_sql)
        result = session.execute(stmt).scalars()
        print('result的值为:',result)
        for obj in result:
            # print(type(obj))
            print(obj.id, obj.name, obj.sal, obj.gender)
        #     print(obj)

4、修改操作

4.1、先查询出来，再修改属性

with sessionmaker(engine).begin() as session:
	emp=session.get(Employee,1)
	emp.sal="60000"

4.2、直接根据主键修改

with sessionmaker(engine).begin() as session:
	stmt=update(Employee).where(Employee.id==5).values(name="库里",sal=66666)
	session.execute(stmt)

4.3、直接根据主机批量修改

with sessionmaker(engine).begin() as session:
	session.execute(update(Employee),[
	            {"id":1,'bonus':800},
	            {"id":5,'bonus':888},
	        ])

5、删除操作

5.1、先查询，再删除

with sessionmaker(engine).begin() as session:
	emp = session.get(Employee, 1)
    session.delete(emp)

5.2、直接删除

with sessionmaker(engine).begin() as session:
	stmt = delete(Employee).where(Employee.id == 8)  # delete from t_emp where id =6
    session.execute(stmt)

6、查询条件

过滤是数据提取的⼀个很重要的功能，以下对⼀些常⽤的过滤条件进⾏解释，并且这些过滤条件都是只能通过where⽅法实现的：

  1. equals：==，或者.is_函数
  2. not equals : !=或者 isnot函数
  3. like & ilike [不区分⼤⼩写]：
  4. 在某个集合中存在，in_函数，或者notin_函数（不存在）
stmt=select(Employee).where(Employee.name=="kobe")
stmt=select(Employee).where(Employee.name.is_(None))  #判断某个字段是否为空

stmt = select(Employee).where(Employee.name != "kobe")
stmt = select(Employee).where(Employee.name.isnot(None))

stmt = select(Employee).where(Employee.name.like("%ko%")) #模糊匹配

stmt = select(Employee).where(Employee.id.in_([1,5]))  # 范围查询

stmt = select(Employee).where(Employee.name.like('%n%'),Employee.sal>20000)  # 模糊查询+多个条件查询+and
stmt = select(Employee).where(and_(Employee.name.like('%n%'),Employee.sal>20000))  # 模糊查询+多个条件查询+and

stmt = select(Employee).where(or_(Employee.name.like('%n%'), Employee.sal > 20000))  # 模糊查询+多个条件查询+or

result = session.execute(stmt).scalars()
for emp in result:
    print(emp)
  1. And多条件组合
stmt = select(Employee).where(Employee.name.like('%n%'),Employee.sal>20000)  # 模糊查询+多个条件查询+and
stmt = select(Employee).where(and_(Employee.name.like('%n%'),Employee.sal>20000))  # 模糊查询+多个条件查询+and
  1. or多条件组合
stmt = select(Employee).where(or_(Employee.name.like('%n%'), Employee.sal > 20000))  # 模糊查询+多个条件查询+or
        
     result = session.execute(stmt).scalars()
     for emp in result:
        print(emp)
  1. 聚合函数

func.count：统计⾏的数量。
func.avg：求平均值。
func.max：求最⼤值。
func.min：求最⼩值。
func.sum：求和。

stmt = select(func.avg(Employee.sal))  # 求平均值
stmt = select(func.sum(Employee.sal))  
res = session.execute(stmt).first()
print(res)
  1. 分⻚查询
esult = session.execute(select(Employee).limit(3)).scalars()
for emp in result:
    print(emp)

result = session.execute(select(Employee).offset(2).limit(3)).scalars()  # 从第3行开始读取3条数据
for emp in result:
    print(emp)
  1. 排序：order_by函数
result = session.execute(select(Employee).order_by(Employee.sal).offset(2).limit(3)).scalars()
for emp in result:
    print(emp)

result = session.execute(select(Employee).order_by(Employee.sal.desc()).offset(2).limit(3)).scalars()
for emp in result:
    print(emp)
  1. 分组查询：group_by和过滤函数having
stmt = select(Employee.gender, func.count(Employee.id)).group_by(Employee.gender)
result = session.execute(stmt).all()
for emp in result:
    print(emp.gender.value,emp.count)
风语者！平时喜欢研究各种技术，目前在从事后端开发工作，热爱生活、热爱工作。

站长推荐