SQLAlchemy 使用经验总结

      class Base(DeclarativeBase): ...


class Company(Base):
    __tablename__ = "companies"
    id: Mapped[int] = mapped_column(primary_key=True) # 主键
    name: Mapped[str] = mapped_column()


class Employee(Base):
    __tablename__ = "employees"
    id: Mapped[int] = mapped_column(primary_key=True) # 主键
    name: Mapped[str] = mapped_column()
    company_id: Mapped[int] = mapped_column(ForeignKey("companies.id")) #  # 外键，关联公司表

    

      class Company(Base):
    __tablename__ = "companies"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    employees: Mapped[List["Employee"]] = relationship() # 关联查询

    

      class Employee(Base):
    __tablename__ = "employees"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    company_id: Mapped[int] = mapped_column(ForeignKey("companies.id"))
    company: Mapped["Company"] = relationship() # 关联查询

    

      class Company(Base):
    __tablename__ = "companies"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    employees: Mapped[list["Employee"]] = relationship()  # 关联查询

class Employee(Base):
    __tablename__ = "employees"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    company_id: Mapped[int] = mapped_column(ForeignKey("companies.id"))
    company: Mapped["Company"] = relationship(back_populates="employees")  # 关联查询，同时定义了反向关系

    

      class TestTable(Base):
    __tablename__ = "test_table"
    default_field: Mapped[int] = mapped_column(server_default="0") # 数据库端默认值

    

      class Test(Base):
    __tablename__ = "test"
    default_field: Mapped[list[str]] = mapped_column(ARRAY(String), default=[])  # 错误使用 default

session = Session()

t1 = Test() 
t1.default_field.append("test") 
print(t1.default_field)  # ['test']

t2 = Test()
print(t2.default_field)  # ['test']，t1 和 t2 共享同一个引用

    

      created_at: Mapped[datetime] = mapped_column(default=func.now())

    

      created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), default=func.now())

    

      default_datetime_now_tz: Mapped[datetime.datetime] = mapped_column(
    TIMESTAMP(timezone=True),
    default_factory=datetime.datetime.now,
)

    

      default_datetime_utcnow_tz: Mapped[datetime.datetime] = mapped_column(
    TIMESTAMP(timezone=True),
    default_factory=lambda: datetime.datetime.now(datetime.UTC),
)

    

      class Base(DeclarativeBase, MappedAsDataclass): ...

    

      class BaseWithAudit(Base):
    __abstract__ = True
    id: Mapped[int] = mapped_column(primary_key=True, init=False)
    created_at: Mapped[datetime.datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now(), init=False)
    updated_at: Mapped[datetime.datetime] = mapped_column(TIMESTAMP(timezone=True), server_default=func.now(), onupdate=func.now(), init=False)

    

      engine = create_engine(os.getenv("DB_URL"))
Session = sessionmaker(bind=engine)

    

      Base.metadata.create_all(engine)

    

      session = Session()
company = session.get(Company, 1) # 根据主键查询获取一个公司对象

    

      stmt = select(Company).where(Company.name == "Google") # 等价于 SELECT * FROM companies WHERE name = "Google"
print(stmt) # 可以打印生成的 SQL 语句。
session.execute(stmt) # 执行查询语句，返回所有符合条件的公司

    

      session.execute(select(Company, Company.name)).all()
# [
#   (<Company id=1>, 'Apple'), 
#   (<Company id=2>, 'Google'), 
#   (<Company id=3>, 'Preferred Networks'),
# ]

    

      session.execute(select(Company)).all()
# [
#   (<Company id=1>,),
#   (<Company id=2>,),
#   (<Company id=3>,),
# ]

    

      session.execute(select(Company.name, Company.id)).scalars().all() # ['Apple', 'Google', 'Preferred Networks']

    

      session.execute(select(Company.name)).scalar() # 'Apple'

    

      session.scalars(select(Company.name)).all() # 等价于 session.execute(select(Company.name)).scalars().all()
session.scalar(select(Company.name)) # 等价于 session.execute(select(Company.name)).scalar()

    

      session = Session()
company = Company(name="Test Company", id=1)
session.add(company)
session.commit() # 此时，执行了 INSERT 语句，将公司添加到数据库中
company.name = "New Company"
session.commit() # 此时，执行了 UPDATE 语句，将公司的名称修改为 New Company
session.delete(company) # 删除公司
session.commit() # 此时，执行了 DELETE 语句，将公司从数据库中删除

    

      session = Session()
alice = Employee(name="Alice", id=1, company_id=1)
session.add(alice)  # 添加 Alice 员工

apple = Company(name="Apple", id=1)
session.add(apple)  # 添加 Apple 公司

bob = Employee(name="Bob", id=2, company=Company(name="Google", id=2))  # 添加 Bob 员工，并直接设定 company 对象
session.add(bob)  # 添加 Bob 员工
session.commit()

    

      apple = Company(name="Apple", id=1)
session.add(apple)  # 添加 Apple 公司
google = Company(name="Google", id=2)
session.add(google)  # 添加 Google 公司
alice = Employee(name="Alice", id=1, company_id=1, company=google)  # id=1，为 Apple 公司，而 company 指定了 Google， Alice 到底是属于 Apple 还是 Google？
session.add(alice)
session.commit()  # 提交事务
assert alice.company_id == 2

    

      session = Session()
session.execute(insert(Company).values(id=1, name="Test Company")) # 执行 INSERT 语句，将公司添加到数据库中

    

      company = Company(name="Test Company", id=1)

session.add(company)
session.commit()

company_1 = session.get(Company, 1)
assert company_1 is company  # True，查询出的对象和之前添加的对象是同一个对象

session.execute(update(Company).where(Company.id == 1).values(name="New Company")) # 修改了公司的名称
session.commit()

assert company.name == "New Company"  # 神奇地，之前存在的对象的值也被修改了

    

      class Company(Base):
    # 省略其他代码
    employees: Mapped[List["Employee"]] = relationship() # 默认 lazy="select"，会延迟加载 

session = Session()
employees = session.scalars(select(Employee)).all()
for employee in employees:
    logger.info("Employee: %s, Company: %s", employee.name, employee.company.name)

    

      stmt = select(Employee).options(selectinload(Employee.company)) # 在查询时，对于 Employee 表的 company 字段使用 selectin 预加载

    

      session = Session()
session.add(Company(name="Test Company", id=1))
session.flush() # 将更改提交到数据库，但不会提交事务

assert session.get(Company, 1).id == 1  # 可以查询到刚刚添加的公司

new_session = Session()
assert new_session.get(Company, 1) is None  # 在别的 session 还查询不到，因为事务没有提交

    

      session = Session()
session.add(Company(name="Test Company", id=1))
session.commit() # 提交事务

new_session = Session()
assert new_session.get(Company, 1).id == 1  # 在别的 session 可以查询到，因为事务已经提交

    

      session = Session()
session.add(Company(name="Test Company", id=1))
session.flush()  # 将更改提交到数据库，但不会提交事务
assert session.get(Company, 1).id == 1  # 此时能查询到
session.rollback()  # 回滚事务，撤销更改
assert session.get(Company, 1) is None  # 此时查询不到

    

      session = Session()
session.add(Company(name="Test Company", id=1))
session.commit()  # 提交事务
assert session.get(Company, 1).id == 1  # 此时能查询到
session.rollback()  # 回滚事务，撤销更改
assert session.get(Company, 1).id == 1  # 此时仍然能查询到，因为事务已经提交，无法回滚

    

      session = Session()
session.add(Company(name="Test Company", id=1))
session.flush()  # 将更改提交到数据库，但不会提交事务
session.close()  # 关闭 session，此时会回滚事务，撤销所有更改

    

      session = Session()
try:
    session.add(Company(name="Test Company", id=1))
    session.flush() # 将更改提交到数据库，但不会提交事务
    raise Exception("Test exception") # 模拟异常
    session.commit() # 提交事务
except Exception:
    session.rollback() # 回滚事务，撤销所有更改

    

      session = Session()
with session.begin():
    session.add(Company(name="Test Company", id=1)) # 添加公司

# 出了 with 语句，事务会自动提交，此外，如果发生异常，事务会自动回滚
new_session = Session()
assert new_session.get(Company, 1) is not None  # 在别的 session 可以查询到，因为事务已经提交

    

          with session.begin():
        session.add(Company(name="Test Company", id=1))
        with session.begin_nested():
            session.add(Employee(name="Test Employee", id=1, company_id=1))
            session.add(Employee(name="Test Employee 2", id=2, company_id=1))

    

      session = sessionmaker(bind=engine)(autoflush=True) # 默认就是开启的，这里显式开启，可以省略
company = Company(name="Google", id=1)
session.add(company)
assert session.get(Company, 1) is not None # 并没有提交更改，也没有提交事务，但是能查询到，因为在查询前进行了自动 flush

    

      session = sessionmaker(bind=engine, autoflush=False)() # 关闭自动 flush
company = Company(name="Google", id=1)
session.add(company)
assert session.get(Company, 1) is None # None，因为没有自动 flush

    

      session = sessionmaker(bind=engine, autoflush=False)()
session.execute(insert(Company).values(id=1, name="Google")) # execute 不需要 flush，会自动提交查询
assert session.get(Company, 1) is not None # 如果使用 execute 进行插入，即使没有自动 flush，仍然能查询到

    

      Session = sessionmaker(bind=engine, expire_on_commit=False)
session = Session()
c = Company(name="Google", id=1)
session.add(c)
session.commit()

# 在另一个 session 中修改了公司的名称
other_session = Session()
other_session.execute(update(Company).where(Company.id == 1).values(name="Meta"))
other_session.commit()

assert c.name == 'Google' # 仍然是 Google，因为没有过期

    

      Session = sessionmaker(bind=engine) # expire_on_commit=True
session = Session()
with session.begin():
    c = Company(name="Google", id=1)
    session.add(c)

with session.begin():
    session.execute(update(Company).where(Company.id == 1).values(name="Meta"))
assert c.name == "Meta"

    

      async_engine = create_async_engine(os.getenv("DB_URL"))
AsyncSession = async_sessionmaker(bind=async_engine)

    

      session = AsyncSession()
async with session.begin():
    session.add(Company(name="Google", id=1))
    await session.commit() # 此时有 I/O 操作，提交事务
    company = await session.get(Company, 1) # 查询 id 为 1 的公司，此时有 I/O 操作
    select_stmt = select(Company).where(Company.name == "Google") # 此时没有 I/O 操作，只是生成了 SQL 语句
    result = await session.execute(select_stmt) # 此时有 I/O 操作，执行查询语句
    companies = result.scalars().all() # 此时没有 I/O 操作，只是处理了查询结果

    

      with Session() as session:
    b = session.get(Employee, 1) # 查询 id 为 1 的员工
    print(b.company.name) # 延迟查询员工的公司名称

    

      async with AsyncSession() as session, session.begin():
    a = await session.get(Employee, 1)
    logger.info(a.company.name) # sqlalchemy.exc.MissingGreenlet

    

      class Base(AsyncAttrs, DeclarativeBase):
    pass

    

      name = await a.awaitable_attrs.company.name

    

      name = await session.run_sync(lambda _: a.company.name)

    

      async def main():
    async with AsyncSession() as session:
        e = await session.scalar(select(Employee).where(Employee.name == "John Doe").options(selectinload(Employee.company)))
        logger.info(f"Employee: {e.company.name}")
        await session.commit()
        await session.refresh(e)
        logger.info(f"Employee: {e.company.name}") # error