创建数据库连接
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 textwith 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()