SQLAlchemy 是 Python 的 SQL 工具包和 ORM 框架
安装
代码语言：javascript复制pip install SQLAlchemy封装
代码语言：javascript复制#path: core/db/sqlite.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
#数据库访问地址
SQLALCHEMY_DATABASE_URL = "sqlite:///./database/app.sqlite3" # SQL
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" # MYSQL
#启动引擎
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
#启动会话
DB_Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 创建会话对象
session = DB_Session()
#数据模型的基类
Base = declarative_base()1.x和2.0 查询语法的区别
https://docs.sqlalchemy.org/en/14/orm/session_basics.html#querying-1-x-style
迁移指南: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html#migration-20-query-usage
1.x 的查询
代码语言：javascript复制# query from a class
results = session.query(User).filter_by(name="ed").all()
# query with multiple classes, returns tuples
results = session.query(User, Address).join("addresses").filter_by(name="ed").all()
# query using orm-columns, also returns tuples
results = session.query(User.name, User.fullname).all()
#总数
session.query(ArticleModel).select_from().count()
session.query(func.count(ArticleModel.id)).scalar()
# 关联查询，获取列表
# .with_entities(ArticleModel.title, CategoryModel.name)
# sqlite不支持concat
# .filter(func.concat(ArticleModel.title, ArticleModel.content).like("�%"))
result = session.query(ArticleModel, CategoryModel)
.join(CategoryModel, CategoryModel.id == ArticleModel.category_id)
.filter(ArticleModel.status == 1, ArticleModel.category_id.in_([1,2]))
.order_by(desc(ArticleModel.create_time))
.offset((page - 1) * max).limit(max)
.all()2.0 的查询
代码语言：javascript复制from sqlalchemy import select
from sqlalchemy.orm import Session
session = Session(engine, future=True)
# query from a class
statement = select(User).filter_by(name="ed")
# list of first element of each row (i.e. User objects)
result = session.execute(statement).scalars().all()
# query with multiple classes
statement = select(User, Address).join("addresses").filter_by(name="ed")
# list of tuples
result = session.execute(statement).all()
# query with ORM columns
statement = select(User.name, User.fullname)
# list of tuples
result = session.execute(statement).all()添加新项或更新现有项
代码语言：javascript复制user1 = User(name="user1")
user2 = User(name="user2")
session.add(user1)
session.add(user2)
session.commit() # write changes to the database要一次向会话添加项目列表，请使用：
代码语言：javascript复制session.add_all([item1, item2, item3])删除
代码语言：javascript复制# mark two objects to be deleted
session.delete(obj1)
session.delete(obj2)
# commit (or flush)
session.commit()练习
代码语言：javascript复制#查询
session.get(ArticleModel, 5)
session.query(ArticleModel).filter(ArticleModel.id == 5).first()
session.query(ArticleModel).filter(ArticleModel.id == id).one()
session.query(ArticleModel).get(5)
#https://docs.sqlalchemy.org/en/14/orm/session_basics.html#update-and-delete-with-arbitrary-where-clause
#1.x的更新
session.query(User).filter(User.name == "squidward").update(
{"name": "spongebob"}, synchronize_session="fetch"
)
#2.0的更新
from sqlalchemy import update
stmt = (
update(User)
.where(User.name == "squidward")
.values(name="spongebob")
.execution_options(synchronize_session="fetch")
)
result = session.execute(stmt)
#获取UPDATE 或 DELETE 受影响的行数，使用
num_rows_matched = result.rowcount
#1.x的删除
session.query(User).filter(User.name == "squidward").delete(synchronize_session="fetch")
#2.0的删除
from sqlalchemy import delete
stmt = (
delete(User)
.where(User.name == "squidward")
.execution_options(synchronize_session="fetch")
)
session.execute(stmt)参考
https://blog.csdn.net/weixin_41085315/article/details/123940220
