sqlalchemy的基本使用

崔哥

发布于 2023-10-23 10:44:11

3630

发布于 2023-10-23 10:44:11

文章被收录于专栏：崔哥的专栏

SQLAlchemy 是 Python 的 SQL 工具包和 ORM 框架

安装

pip install SQLAlchemy

封装

#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 的查询

# 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("%cc%")) \
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 的查询

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()

添加新项或更新现有项

user1 = User(name="user1")
user2 = User(name="user2")
session.add(user1)
session.add(user2)

session.commit()  # write changes to the database

要一次向会话添加项目列表，请使用：

session.add_all([item1, item2, item3])

删除

# mark two objects to be deleted
session.delete(obj1)
session.delete(obj2)

# commit (or flush)
session.commit()

练习

#查询
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)