当前位置：首页 > news >正文

sqlalchemy orm查询语法与多对多外键关系relationship使用

news 来源：原创 2025/5/15 4:53:02

创建数据库连接

SQLAlchemy连接到数据库，我们需要创建一个Engine对象。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

# 创建基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# 创建数据库引擎和会话
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

查询语句

# 查询

查询所有用户数据
session.query(User).all() ==> [user1,user2]

查询有多少个用户
session.query(User).count()

查询第1个用户
session.query(User).first()

查询id为4的用户[3种方式]
session.query(User).get(4) (这里面只能是数字 或 ‘4’) 查询不到不会报错
session.query(User).filter_by(id = 4).first() （一个等号）查询不到不会报错
session.query(User).filter(User.id == 4).first()  查询不到不会报错

filter方法实现的：
1. equals ： ==
    news= session.query(News).filter(News.title == "title1").first()
    query里面是列查询，字段，传入模型名表示全字段 。
    filter是行查询，没有filter表示全表查询
2. not equals  :  !=
    query(User).filter(User.name != 'ed')

3. like[不区分大小写]  & ilike[区分大小写]：
    query(User).filter(User.name.like('%ed%··')) 
 query(User).filter(User.name.like('__ed%__')) _表示站位符

4. in：
    query(User).filter(User.name.in_(['ed','wendy','jack']))

5. not in：
    query(User).filter(~User.name.in_(['ed','wendy','jack']))

6.  is null：
    query(User).filter(User.name==None)
    # 或者是
    query(User).filter(User.name.is_(None))

7. is not null:
    query(User).filter(User.name != None)
    # 或者是
    query(User).filter(User.name.isnot(None))

8. and：
    query(User).filter(and_(User.name=='ed',User.fullname=='Ed Jones'))
    # 或者是传递多个参数
    query(User).filter(User.name=='ed',User.fullname=='Ed Jones')
    # 或者是通过多次filter操作
    query(User).filter(User.name=='ed').filter(User.fullname=='Ed Jones')
9. or：
    query(User).filter(or_(User.name=='ed',User.name=='wendy'))

10. order_by
查询所有用户数据，并以邮箱排序
session.query(User).order_by(User.email).all()
session.query(User).order_by(User.email.desc()).all()

11. 查询名字结尾字符为g的所有数据[开始/包含]
session.query(User).filter(User.name.endswith('g')).all()
session.query(User).filter(User.name.startswith('g')).all()
session.query(User).filter(User.name.contains('g')).all()


每页3个，查询第2页的数据
paginate = session.query(User).paginate(page, per_page,Error_out)
paginate = session.query(User).paginate(2,3,False)
page: 哪一个页
per_page: 每页多少条数据
Error_out: False 查不到不报错

paginate .pages: 共有多少页
paginate .items: 当前页数的所有对象
paginate .page: 当前页
sqlalchemy查询常见的过滤方式
filter过滤条件：
过滤是数据提取的一个很重要的功能，以下对一些常用的过滤条件进行解释，并且这些过滤条件都是只能通过

如果想要查看orm底层转换的sql语句，可以在filter方法后面不要再执行任何方法直接打印就可以看到了。比如：
news = session.query(News).filter(or_(News.title=='abc',News.content=='abc'))
print(news)

增加

# 增加
user = User(name='laowang')
session.add(user)
session.commit()
# 新增多条数据
user_obj1 = User(name="bigberg", passwd="twgdh123")
user_obj2 = User(name="someone", passwd="twgdh123")
Session.add_all([user_obj1,user_obj2])
# 批量新增方式一
session.bulk_save_objects(
    [
        User(username=username + str(i), password=password)
        for i in range(10000)
    ]
)
session.commit()
# 批量新增方式二
session.bulk_insert_mappings(
    User,
    [
        dict(username="NAME INSERT " + str(i), password=password)
        for i in range(10000)
    ]
)
session.commit()
# 批量新增方式三
session.execute(
    User.__table__.insert(),
    [{"username": 'Execute NAME ' + str(i), "password": password} for i in range(10000)]
)
db.session.commit()

修改

# 修改
session.query(AiQuestionSortSse).filter(AiQuestionSortSse.id.in_(id_list)).update({"ptime": datetime.now()})
session.commit()

删除

# 删除
session.query(TaskSubOriginQuestion).where(TaskSubOriginQuestion.id.in_(ids)).delete()
session.commit()

SQLAlchemy：执行原生 SQL 语句

# 方式一使用engine
# 创建 Engine 后，使用 connect() 方法获取一个连接（Connection）对象，它提供了 execute() 方法来运# 行 SQL 语句：
from sqlalchemy import text

with engine.connect() as connection:
    result = connection.execute(text("SELECT * FROM users WHERE id=:user_id"), {"user_id": 1})
    for row in result:
        print(row)
# 方式二使用 session
result = session.execute(text("SELECT * FROM users WHERE id=:user_id"), {"user_id": 1})
for row in result:
    print(row)

session.close()

# 更新和插入
session.execute(text("UPDATE users SET name=:name WHERE id=:user_id"), {"name": "Alice", "user_id": 1})
session.commit()
session.execute(text("DELETE FROM users WHERE id=:user_id"), {"user_id": 1})
session.commit()