SqlAlchemy学习总结

1 安装

pip install sqlalchemy

2 创建Engine

2.1 数据库连接协议：

SQLite: sqlite:///./mydb.db  "sqlite:///:memory:"(内存型)
MySQL: mysql://username:password@hostname/database_name
PostgreSQL: postgresql://username:password@hostname/database_name
SQL Server: mssql+pyodbc://username:password@hostname/database_name?driver=ODBC+Driver+17+for+SQL+Server
Oracle: oracle+cx_oracle://username:password@hostname:port/service_name
MariaDB: mariadb://username:password@hostname/database_name
MongoDB: mongodb+srv://username:password@cluster_name/database_name

2.2 示例语法：

from sqlalchemy import create_engine
# echo: python日志记录器输出
engine = create_engine("sqlite+pysqlite:///./mydb.db:", echo=True)

3 创建Connection

# from sqlalchemy import text
# 方式1：
with engine.connect() as conn:
    result = conn.execute(text("select 'hello world'"))
    print(result.all())
    
# 方式2：
with session.connection() as conn:
    result = conn.execute(text("select 'hello world'"))
    print(result.all())

4 读取表数据

# 以行读取
from sqlalchemy import text

with engine.connect() as conn:
	result = conn.execute(text("SELECT x, y FROM some_table"))

# 元组赋值
for x, y in result:
...

# 整数索引
for row in result:
x = row[0]
y = row[1]

# 属性名称
for row in result:
print(f"x: {row.x}  y: {row.y}")

# 映射访问：
for dict_row in result.mappings():
x = dict_row["x"]
y = dict_row["y"]

5 execute传参

# ":num": 以"冒号"开头来定义占位参数，然后方法的第二个参数以“{}”的形式进行传参
result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :num"), {"num": 2})

# 多个参数：以“[{},{},...]”形式传参
conn.execute(
  text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
  [{"x": 11, "y": 12}, {"x": 13, "y": 14}],
)

6 MetaData

6.1  应用场景：

主要和表结构相关：reflect(表结构反射)、orm

# reflect：把数据库中的表结构信息反射到MetaData对象中
# orm：通过模型类创建表到数据库

6.2 reflect：

engine = create_engine('sqlite:///mydb.db')
# 创建 Metadata 对象
metadata_obj = MetaData(bind=engine)
# 从数据库中读取表格结构信息并反映到 Metadata 对象中
metadata_obj.reflect()
# 打印反映后的表格信息
for table in metadata_obj.tables.values():
    print("Table Name:", table.name)
    print("Columns:")
    for column in table.columns:
        print("\tColumn Name:", column.name)
        print("\tData Type:", column.type)
  
 
# 为Table设置MetaData        
from sqlalchemy import Table, Column, Integer, String
user_table = Table(
    "user_account",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("name", String(30)),
    Column("fullname", String),
    extend_existing=True
)
# user_table.primary_key
# user_table.c.name
# user_table.c.keys()

 6.3 表创建：

metadata_obj.create_all(engine)

# 未绑定engine：sqlalchemy.exc.UnboundExecutionError: MetaData object is not bound to an Engine or Connection.  Execution can not proceed without a database to execute against.
metadata_obj = MetaData(bind=engine)

# 表已存在：sqlalchemy.exc.InvalidRequestError: Table 'user' is already defined for this MetaData instance.  Specify 'extend_existing=True' to redefine options and columns on an existing Table object.
user_table = Table(
     "user",
     metadata_obj,
     ...,
     extend_existing=True
)

7 声明式orm

7.1 1.x版本：

# 注：2.0版本语法参考sqlalchemy(2.0)小结
from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
	pass

class User(Base):
  __tablename__ = "user_account"

     id = Column(Integer, primary_key=True)
     name = Column(String(30), nullable=False)
     fullname = Column(String)
  addresses = relationship("Address", back_populates="user")

class Address(Base):
  __tablename__ = "address"

     id = Column(Integer, primary_key=True)
     email_address = Column(String, nullable=False)
     user_id = Column(ForeignKey("user_account.id"), nullable=False)
  user = relationship("User", back_populates="addresses")

7.2 创建表：

Base.metadata.create_all(engine)

8 创建对象并持久化

8.1 创建会话：

# 以下两种方式创建会话本质没区别
# 方式1：
from sqlalchemy.orm import sessionmaker
session = sessionmaker(bind=engine)

# 方式2：
from sqlalchemy.orm import Session
session = Session(bind=engine)

8.2 持久化：

from sqlalchemy.orm import Session

with Session(engine) as session:
spongebob = User(
  name="spongebob",
  fullname="Spongebob Squarepants",
  addresses=[Address(email_address="spongebob@sqlalchemy.org")],
)
sandy = User(
  name="sandy",
  fullname="Sandy Cheeks",
  addresses=[
      Address(email_address="sandy@sqlalchemy.org"),
      Address(email_address="sandy@squirrelpower.org"),
  ],
)
patrick = User(name="patrick", fullname="Patrick Star")
session.add_all([spongebob, sandy, patrick])
session.commit()

9 CRUD(sqlalchemy.x)

9.1 简要：

在sqlalchemy包中，已经封装了crud的方法，可以很方便调用，从而不去写sql。
注意仅是将sql语句的方法化，执行这些方法并不是“执行sql”。
from sqlalchemy import insert,delete,update,select

# 示例model实例，供下面参考（采用1.x版本语法）
class User(Base):
     __tablename__ = "users"
     id = Column(Integer,primary_key=True)
     name = Column(String(30))
     fullname = Column(String)

9.2 crud：

# insert
stmt = (
    insert(User).
    values(name="jd",fullname="joden")
)

# delete
stmt = (
    delete(User).
    where(User.name=="jd")
)

# update
stmt = (
    update(User).
    where(User.name=="u1").
    values(name="j1")
)

# select
stmt = (
    select(User).
    where(name=="j1")
)

9.3 执行sql：

# 正如“简要”中提到，执行上述方法并没有执行sql（产生数据库变化）
# 如下语句将执行sql:

insert_stmt = (
     insert(User).
     values(name="jd",fullname="joden")
)

from sqlalchemy import create_engine
# url = "sqlite:///./t1.db"
engine = create_engine(url)
with engine.connect() as conn:
     conn.execute(insert_stmt)
     conn.commit()

 
 
# 另外地，如果是select(查询操作)，也可以使用session.scalar(stmt)/scalars(stmt)来获取以实例为标量的查询集
from sqlalchemy.orm import Session
session = Session(engine)

res = session.scalars(stmt)
for row in res:
    print(row.name,row.fullname)

10 CRUD(session.x)

10.1 简要：

        与sqlalchemy.x中的不同的是，session.x中的crud方法会在当前会话中执行，在会话管理中作为数据库待定操作集，但是仍然需要我们去做会话(事务)提交才能真正做出数据库操作。

        所以说简单较少的数据库操作可以简单通过sqlalchemy.x中方法，而较多的数据库操作集则用session.x中方法比较好。

10.2 方法：  

10.3 简单示例：  

# select by id
sandy = session.get(User, 2)
# select
session.query(User).where(name="joden")
session.query(User.name,User.age).where(name="joden")

# add 
u1 = User(name="j1",fullname="joden1")
u2 = User(name="j2",fullname="joden2")
# session.add(u1)
session.add_all([u1,u2])

# delete
session.delete(u1)

# update
u1.name = "jd1"

# session commit
session.commit()

10.4 session.flush：

# 应用场景
 （1）获取新创建对象的主键值： 如果你需要在创建对象后立即获取其分配的主键值，那么在插入操作后执行 flush() 方法是必要的。否则，新创建的对象的主键值将不会在对象上更新，直到事务提交后才会更新。
 （2）处理数据库的唯一约束和触发器： 在某些情况下，数据库中的唯一约束或触发器可能会影响到插入或更新操作。执行 flush() 可以确保你及时地发现并处理任何由于约束或触发器引发的异常。
 （3）在需要确保数据一致性的操作中： 在一些需要确保数据一致性的复杂操作中，flush() 可以帮助你将操作分阶段提交到数据库，确保在整个操作过程中数据库的状态保持一致。

u4 = User(name="j4",fullname="joden4")
session.add(u4)
session.flush()
# print(u4.id)
session.commit()

11 execute

在sqlalchemy中使用execute(stmt)来执行sql。
要注意的是，可通过不同的实例来调用execute，下面我们将来了解几者的区别。

（1）engine.connect().execute(stmt)
（2）session.execute(stmt) 
（3）session.connection().execute(stmt) # 本质上是engine.connect().execute(stmt)

所以区别仅在于Connection对象还是Session对象执行，很明显通过Connection对象方式需要自己创建管理连接，而Session对象方式直接执行则不需要(但是需要关闭当前“会话”)。

12 session状态

13 别名

13.1 字段别名：

for row in session.query(User.name.label("name_label")).all():
	print(row.name_label)

13.2 表别名：

from sqlalchemy.orm import aliased

user_alias = aliased(User, name="user_alias")

for row in session.query(user_alias).all():
	print(row.name)

14 过滤

14.1 filter_by:

for (name,) in session.query(User.name).filter_by(fullname="joden"):
	print(name)

14.2 filter：

# 较为灵活，支持python逻辑运算符
for (name,) in session.query(User.name).filter(User.fullname == "joden"):
	print(name)


# 常用操作
query.filter(User.name == "ed")
query.filter(User.name != "ed")
query.filter(User.name.like('%ed%'))
query.filter(User.name.in_(["ed", "wendy", "jack"]))
query.filter(User.name.in_(session.query(User.name))

14.3 链式调用：

# 无论是filter还是filter_by都可以链式调用
session.query(User).filter(User.name == "j1").filter(User.fullname == "joden1")

14.5 and_：

from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# 等价下面两种方式
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')

14.6 text：

from sqlalchemy import text

# 基础语法
session.query(User).filter(text("id<224"))
# 使用占位参数
session.query(User).filter(text("id<:value and name=:name")).params(value=3, name="jd")

15 join查询

15.1 隐式内连接：

session.query(User, Address)
 .filter(User.id == Address.user_id)
 .filter(Address.email_address == "jack@google.com")
 .all()

15.2 显示内连接：

session.query(User)
	.join(Address)
	.filter(Address.email_address == "jack@google.com")

15.3 外连接：

# LEFT OUTER JOIN
query.outerjoin(User.addresses)  
query.outerjoin(Address)  

# select_from 控制左右外连接
query = session.query(User, Address).select_from(Address).join(User)

16 表关系

16.1 一对多：

### 方式一：
# (1) 在两表中各定义
relationship(
     association_table_name,
     back_populates=association_table_attr 	# 在关联表中定义的关联属性
)
# (2) 在子表中定义外键属性(字段)
parent_table_id = Column(ForeignKey("parent_table_name.id"))


### 方式二：
# (1) 在一表中定义
relationship(
     association_table_name,
     backref=association_table_attr
)
# (2) 在子表中定义外键属性(字段)
parent_table_id = Column(ForeignKey("parent_table_name.id"))



### 示例代码
class TB1(Base):
     __tablename__ = 'tb1'

     id = Column(Integer,primary_key=True)
     name = Column(String(20))
     t2 = relationship("TB2",backref="t1")

class TB2(Base):
     __tablename__ = 'tb2'

     id = Column(Integer,primary_key=True)
     name = Column(String(20))
     t1_id = Column(ForeignKey("tb1.id"))


### 一对多关联查询
session.query(TB1).first().t2

16.2 多对多：

### 方式一：
association_table = Table(
    "association_table",
    Base.metadata,
    Column("left_id", ForeignKey("left_table.id"), primary_key=True),
    Column("right_id", ForeignKey("right_table.id"), primary_key=True),
)

class Parent(Base):
    __tablename__ = "left_table"
    id = Column(Integer, primary_key=True)
    children = relationship(
        "Child", secondary=association_table, back_populates="parents"
    )

class Child(Base):
    __tablename__ = "right_table"
    id = Column(Integer, primary_key=True)
    parents = relationship(
        "Parent", secondary=association_table, back_populates="children"
    )


### 方式二：
association_table = Table(
    "association_table",
    Base.metadata,
    Column("left_id", ForeignKey("left_table.id"), primary_key=True),
    Column("right_id", ForeignKey("right_table.id"), primary_key=True),
)

class Parent(Base):
    __tablename__ = "left_table"
    id = Column(Integer, primary_key=True)
    children = relationship("Child", secondary=association_table, backref="parents")

class Child(Base):
    __tablename__ = "right_table"
    id = Column(Integer, primary_key=True)

16.3 一对一：

# 一对一通过添加uselist=False参数即可
class Parent(Base):
     __tablename__ = "parent_table"
     id = Column(Integer, primary_key=True)
     child = relationship("Child", back_populates="parent", uselist=False)

class Child(Base):
     __tablename__ = "child_table"
     id = Column(Integer, primary_key=True)
     parent_id = Column(Integer, ForeignKey("parent_table.id"))

     # many-to-one side remains, see tip below
     parent = relationship("Parent", back_populates="child")

17 backref和back_populates

17.1 区别：

        backref 是 back_populates 的一种简化形式。backref 允许你在定义关系的同时，在另一个表格中自动创建一个反向引用属性，而不需要显式地在另一个表格的 relationship() 中设置 back_populates。

17.2 back_populates：  

class Parent(Base):
    __tablename__ = 'parents'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")

class Child(Base):
    __tablename__ = 'children'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parents.id'))
    parent = relationship("Parent", back_populates="children")

17.3 backref：  

class Parent(Base):
     __tablename__ = 'parents'
     id = Column(Integer, primary_key=True)
     children = relationship("Child", backref="parent")

class Child(Base):
     __tablename__ = 'children'
     id = Column(Integer, primary_key=True)
     parent_id = Column(Integer, ForeignKey('parents.id'))

18 session.query

18.1 执行简单查询：

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

# 查询所有记录
result = session.query(User).all()


for instance in session.query(User).order_by(User.id):
	print(instance.name, instance.fullname)

18.2 常用查询：

# 过滤查询： 你可以使用 `filter()` 方法对查询结果进行过滤。
result = session.query(User)
    .filter(User.age > 25)
    .all()

# 排序查询： 使用 `order_by()` 方法对查询结果进行排序。
result = session.query(User)
    .order_by(User.age.desc())
    .all()

# 连接查询： 你可以执行连接查询来检索相关联的对象。
result = session.query(User, Address)
    .join(Address)
    .filter(User.id == Address.user_id)
    .all()

# 聚合查询： 使用 `func` 函数进行聚合查询。
from sqlalchemy import func
result = session.query(func.count(User.id)).scalar()

# 子查询： 你可以在查询中使用子查询来执行复杂的操作。
subquery = session.query(Address.user_id).filter(Address.city == 'New York')
result = session.query(User).filter(User.id.in_(subquery)).all()

19 sqlalchemy(2.0)

19.1 声明映射类：

# 不同：使用python类型简化类型声明
# 注意：对于字符串类型字段，如果在mysql中创建则必须要给定长度(sqlite、PostgreSQL中则不用)

from typing import List
from typing import Optional

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
 	pass

class User(Base):
     __tablename__ = "user_account"
     id: Mapped[int] = mapped_column(primary_key=True)
     name: Mapped[str] = mapped_column(String(30))
     fullname: Mapped[Optional[str]]  # 必要时要补全
     addresses: Mapped[List["Address"]] = relationship(back_populates="user")

     def __repr__(self) -> str:
         return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"


class Address(Base):
     __tablename__ = "address"
     id: Mapped[int] = mapped_column(primary_key=True)
     email_address: Mapped[str] # 必要时要补全
     user_id = mapped_column(ForeignKey("user_account.id"))
     user: Mapped[User] = relationship(back_populates="addresses")

     def __repr__(self) -> str:
         return f"Address(id={self.id!r}, email_address={self.email_address!r})"

19.2 抽离重复类型字段：

# 比如主键用到的比较多且类型一致：整型、主键
# 2.0后可以使用类型注解的形式来对重复字段类型进行抽离定义，从而来简化字段声明

from typing_extensions import Annotated
from sqlalchemy.orm import Mapped

str50 =  Annotated[str, mapped_column(String(50))]
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]


class User(Base):
 	name:Mapped[str50]

20 数据迁移(alembic)

20.1 安装迁移依赖：

pip install alembic

20.2 迁移须知：

1. 使用alembic迁移配置需要导入基类Base，此时只能通过declarative_base()来创建，而不能通过DeclarativeBase来创建

2. 对于数据库迁移，我们始终需要两个信息：
    (1) 要迁移到的数据库
    (2) 要迁移的表
    
3. 所以使用alembic初始化我们需要配置以上的两个信息：
    初始化命令(项目根目录下执行，一般与app目录同级)：alembic init alembic
    
4. 配置两个信息：
    (1) 数据库信息：alembic.ini中的sqlalchemy.url进行配置
    (2) 要迁移的表：因为要迁移的表都与基类Base所关联，并且sqlalchemy创建表是通过Base.metadata来创建表到数据库，所以alembic要求我们配置Base.metadata，在alembic/env.py中的target_metadata进行配置

20.3 数据迁移：

# 安装好依赖，并且定义好所需的表模型后，我们将开始迁移
# 1. 初始化配置：alembic init alembic
# 2. 生成迁移文件：alembic revision --autogenerate -m [message]（该message是迁移备注信息，可以自己定义）
# 3. 开始迁移：alembic upgrade head

# 注：我们可以关注每一步迁移命令产生的变化或文件，从而有利于我们进一步了解迁移的本质

21 英语词汇

Declare：声明        typically：通常         conjunction：结合        persist：持久化         

emit：发出        abbreviated：缩写     particular：特定