自动 ID、None 默认值和数据刷新¶
在上一章中，我们了解了如何使用 SQLModel 将行添加到数据库中。
现在我们来谈谈为什么数据库上的 id 字段不能为 NULL，因为它是一个主键，我们使用 Field(primary_key=True) 来声明它。
但是同一个 id 字段在 Python 代码中实际上可以为 None，所以我们使用 int | None 声明类型，并将默认值设置为 Field(default=None)
# Code above omitted 👆
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
接下来，我将向您展示更多关于数据库和 Python 代码之间数据同步的信息。
我们什么时候才能从数据库中获得该 id 字段的实际 int 值？让我们看看所有这些。👇
创建一个新的 Hero 实例¶
当我们创建一个新的 Hero 实例时，我们不设置 id
# Code above omitted 👆
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
int | None 如何提供帮助¶
因为我们没有设置 id，它会采用我们在 Field(default=None) 中设置的 Python 默认值 None。
这就是我们使用 int | None 并将默认值设置为 None 的唯一原因。
因为在代码的这一点上，在与数据库交互之前，Python 值实际上可能是 None。
如果我们假设 id 总是一个 int 并且没有 int | None 添加类型注解，我们可能会写出损坏的代码，例如
next_hero_id = hero_1.id + 1
如果我们在将英雄保存到数据库之前运行此代码，并且 hero_1.id 仍然是 None，我们将收到一个类似以下的错误
TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'
但通过使用 int | None 声明它，编辑器将通过向我们显示警告来帮助我们避免编写损坏的代码，该警告告诉我们如果 hero_1.id 为 None，代码可能无效。🔍
打印默认的 id 值¶
我们可以通过在将英雄添加到数据库之前打印它们来确认这一点
# Code above omitted 👆
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
这将输出
$ python app.py
// Output above omitted 👆
Before interacting with the database
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48
请注意，它们都具有 id=None。
这是我们在 Hero 模型类中定义的默认值。
当我们把这些对象 add 到会话中时会发生什么？
将对象添加到会话¶
我们将 Hero 实例对象添加到会话后，ID 仍然是 None。
我们可以通过使用 with 块创建会话并添加对象，然后再次打印它们来验证
# Code above omitted 👆
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
这会再次将对象的 id 输出为 None
$ python app.py
// Output above omitted 👆
After adding to the session
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48
正如我们之前看到的，会话很智能，它不会在我们每次准备更改某些内容时都与数据库通信，只有在我们准备好并告诉它 commit 更改后，它才会向数据库发送所有 SQL 以存储数据。
将更改提交到数据库¶
然后我们可以 commit 会话中的更改，然后再次打印
# Code above omitted 👆
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
现在，发生了一些意想不到的事情，看看输出，Hero 实例对象似乎根本没有数据
$ python app.py
// Output above omitted 👆
// Here the engine talks to the database, the SQL part
INFO Engine BEGIN (implicit)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [generated in 0.00018s] ('Deadpond', 'Dive Wilson', None)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [cached since 0.0008968s ago] ('Spider-Boy', 'Pedro Parqueador', None)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [cached since 0.001143s ago] ('Rusty-Man', 'Tommy Sharp', 48)
INFO Engine COMMIT
// And now our prints
After committing the session
Hero 1:
Hero 2:
Hero 3:
// What is happening here? 😱
发生的情况是 SQLModel（实际上是 SQLAlchemy）在内部将这些对象标记为“已过期”，它们没有其数据的最新版本。这是因为我们可能在数据库中更新了一些字段，例如，假设一个字段 updated_at: datetime 在我们保存更改时自动更新。
同样，其他值可能已更改，因此会话为了确保安全的选择是只在内部将对象标记为已过期。
然后，下次我们访问每个属性时，例如使用
current_hero_name = hero_1.name
...SQLModel（实际上是 SQLAlchemy）将确保联系数据库并获取数据的最新版本，更新我们对象中的 name 字段，然后使其可用于 Python 表达式的其余部分。在上面的示例中，此时，Python 将能够继续执行并使用该 hero_1.name 值（刚刚更新）将其放入变量 current_hero_name 中。
所有这些都自动在幕后发生。✨
这是我们示例中有趣而奇怪的地方
print("Hero 1:", hero_1)
我们没有访问对象的属性，例如 hero.name。我们只访问了整个对象并打印了它，因此 SQLAlchemy 无法知道我们想要访问此对象的数据。
打印单个字段¶
为了确认和理解在访问属性时数据自动过期和刷新的工作原理，我们可以打印一些单个字段（实例属性）
# Code above omitted 👆
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
现在我们实际上正在访问属性，因为我们没有打印整个对象 hero_1
print("Hero 1:", hero_1)
...我们现在正在打印 hero.id 中的 id 属性
print("Hero 1 ID:", hero_1.id)
通过访问该属性，这会触发 SQLModel（实际上是 SQLAlchemy）在底层完成大量工作，以从数据库中刷新数据，将其设置在对象的 id 属性中，并使其可用于 Python 表达式（在本例中仅用于打印）。
让我们看看它是如何工作的
$ python app.py
// Output above omitted 👆
// After committing, the objects are expired and have no values
After committing the session
Hero 1:
Hero 2:
Hero 3:
// Now we will access an attribute like the ID, this is the first print
After committing the session, show IDs
// Notice that before printing the first ID, the Session makes the Engine go to the database to refresh the data 🤓
INFO Engine BEGIN (implicit)
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [generated in 0.00017s] (1,)
// Here's our first print, now we have the database-generated ID
Hero 1 ID: 1
// Before the next print, refresh the data for the second object
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.001245s ago] (2,)
// Here's our print for the second hero with its auto-generated ID
Hero 2 ID: 2
// Before the third print, refresh its data
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.002215s ago] (3,)
// And here's our print for the third hero
Hero 3 ID: 3
// What if we print another attribute like the name?
After committing the session, show names
Hero 1 name: Deadpond
Hero 2 name: Spider-Boy
Hero 3 name: Rusty-Man
// Because the Session already refreshed these objects with all their data and the session knows they are not expired, it doesn't have to go again to the database for the names 🤓
显式刷新对象¶
您刚刚了解到会话如何在幕后，作为副作用，在您访问属性时自动刷新数据。
但是如果您想显式刷新数据怎么办？
您也可以使用 session.refresh(object) 来做到这一点
# Code above omitted 👆
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
当 Python 执行此代码时
session.refresh(hero_1)
...会话会使用引擎与数据库通信，以获取此对象 hero_1 的最新数据，然后会话将数据放入 hero_1 对象并将其标记为“新鲜”或“未过期”。
以下是输出的样子
$ python app.py
// Output above omitted 👆
// The first refresh
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [generated in 0.00024s] (1,)
// The second refresh
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.001487s ago] (2,)
// The third refresh
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.002377s ago] (3,)
// Now print the data, as it's already refreshed there's no need for the Session to go and refresh it again
After refreshing the heroes
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson'
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador'
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
这可能很有用，例如，如果您正在构建一个用于创建英雄的 Web API。一旦英雄创建了一些数据，您就将其返回给客户端。
您不希望返回一个看起来为空的对象，因为没有触发自动刷新数据的魔术。
在这种情况下，在通过会话将对象提交到数据库后，您可以刷新它，然后将其返回给客户端。这将确保对象具有其新鲜数据。
关闭会话后打印数据¶
现在，作为最后的实验，我们还可以在会话关闭后打印数据。
这里没有什么惊喜，它仍然有效
# Code above omitted 👆
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
print("Before interacting with the database")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
print("After adding to the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
session.commit()
print("After committing the session")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After committing the session, show IDs")
print("Hero 1 ID:", hero_1.id)
print("Hero 2 ID:", hero_2.id)
print("Hero 3 ID:", hero_3.id)
print("After committing the session, show names")
print("Hero 1 name:", hero_1.name)
print("Hero 2 name:", hero_2.name)
print("Hero 3 name:", hero_3.name)
session.refresh(hero_1)
session.refresh(hero_2)
session.refresh(hero_3)
print("After refreshing the heroes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
print("After the session closes")
print("Hero 1:", hero_1)
print("Hero 2:", hero_2)
print("Hero 3:", hero_3)
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
输出再次显示相同的数据
$ python app.py
// Output above omitted 👆
// By finishing the with block, the Session is closed, including a rollback of any pending transaction that could have been there and was not committed
INFO Engine ROLLBACK
// Then we print the data, it works normally
After the session closes
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson'
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador'
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
回顾所有代码¶
现在让我们再次回顾所有这些代码。
提示
每个编号的气泡都显示每行将在输出中打印什么。
并且由于我们使用 echo=True 创建了引擎，我们可以在每个步骤中看到执行的 SQL 语句。
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (1)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador") # (2)!
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48) # (3)!
print("Before interacting with the database") # (4)!
print("Hero 1:", hero_1) # (5)!
print("Hero 2:", hero_2) # (6)!
print("Hero 3:", hero_3) # (7)!
with Session(engine) as session: # (8)!
session.add(hero_1) # (9)!
session.add(hero_2) # (10)!
session.add(hero_3) # (11)!
print("After adding to the session") # (12)!
print("Hero 1:", hero_1) # (13)!
print("Hero 2:", hero_2) # (14)!
print("Hero 3:", hero_3) # (15)!
session.commit() # (16)!
print("After committing the session") # (17)!
print("Hero 1:", hero_1) # (18)!
print("Hero 2:", hero_2) # (19)!
print("Hero 3:", hero_3) # (20)!
print("After committing the session, show IDs") # (21)!
print("Hero 1 ID:", hero_1.id) # (22)!
print("Hero 2 ID:", hero_2.id) # (23)!
print("Hero 3 ID:", hero_3.id) # (24)!
print("After committing the session, show names") # (25)!
print("Hero 1 name:", hero_1.name) # (26)!
print("Hero 2 name:", hero_2.name) # (27)!
print("Hero 3 name:", hero_3.name) # (28)!
session.refresh(hero_1) # (29)!
session.refresh(hero_2) # (30)!
session.refresh(hero_3) # (31)!
print("After refreshing the heroes") # (32)!
print("Hero 1:", hero_1) # (33)!
print("Hero 2:", hero_2) # (34)!
print("Hero 3:", hero_3) # (35)!
# (36)!
print("After the session closes") # (37)!
print("Hero 1:", hero_1) # (38)!
print("Hero 2:", hero_2) # (39)!
print("Hero 3:", hero_3) # (40)!
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
-
创建
hero_1。不产生任何输出.
-
创建
hero_2。不产生任何输出.
-
创建
hero_3。不产生任何输出.
-
打印行
"Before interacting with the database"。生成输出
Before interacting with the database -
在与数据库交互之前打印
hero_1。生成输出
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None -
在与数据库交互之前打印
hero_2。生成输出
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None -
在与数据库交互之前打印
hero_3。生成输出
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48 -
在
with块中创建Session。不产生任何输出.
-
将
hero_1添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
将
hero_2添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
将
hero_3添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
打印行
"After adding to the session"。生成输出
After adding to the session -
在添加到会话后打印
hero_1。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None -
在添加到会话后打印
hero_2。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None -
在添加到会话后打印
hero_3。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48 -
commit会话。这将保存所有数据到数据库。会话将使用引擎运行大量 SQL。
生成输出
INFO Engine BEGIN (implicit) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [generated in 0.00018s] ('Deadpond', 'Dive Wilson', None) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [cached since 0.0008968s ago] ('Spider-Boy', 'Pedro Parqueador', None) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [cached since 0.001143s ago] ('Rusty-Man', 'Tommy Sharp', 48) INFO Engine COMMIT -
打印行
"After committing the session"。生成输出
After committing the session -
在提交会话后打印
hero_1。hero_1现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 1: -
在提交会话后打印
hero_2。hero_2现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 2: -
在提交会话后打印
hero_3。hero_3现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 3: -
打印行
"After committing the session, show IDs"。生成输出
After committing the session, show IDs -
打印
hero_1.id。这里发生了很多事情。因为我们正在访问
hero_1的属性id，SQLModel（实际上是 SQLAlchemy）可以检测到我们正在尝试访问hero_1的数据。然后它检测到
hero_1当前与一个会话关联（因为我们将其添加到会话并提交了它），并且它被标记为已过期。然后通过会话，它使用引擎执行所有 SQL，从数据库中获取此对象的数据。
接下来，它使用新数据更新对象，并在内部将其标记为“新鲜”或“未过期”。
最后，它使 ID 值可用于 Python 表达式的其余部分。在这种情况下，Python 表达式只是打印 ID。
生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00017s] (1,) Hero 1 ID: 1 -
打印
hero_2.id。这里发生了很多事情，所有与第 22 点相同的事情，但针对的是
hero_2对象。生成输出
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001245s ago] (2,) Hero 2 ID: 2 -
打印
hero_3.id。这里发生了很多事情，所有与第 22 点相同的事情，但针对的是
hero_3对象。生成输出
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.002215s ago] (3,) Hero 3 ID: 3 -
打印行
"After committing the session, show names"。生成输出
After committing the session, show names -
打印
hero_1.name。因为
hero_1仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 1 name: Deadpond -
打印
hero_2.name。因为
hero_2仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 2 name: Spider-Boy -
打印
hero_3.name。因为
hero_3仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 3 name: Rusty-Man -
显式刷新
hero_1对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_1对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00024s] (1,) -
显式刷新
hero_2对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_2对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001487s ago] (2,) -
显式刷新
hero_3对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_3对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.002377s ago] (3,) -
打印行
"After refreshing the heroes"。生成输出
After refreshing the heroes -
打印
hero_1。信息
即使
hero_1不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_1是新鲜的，它拥有所有可用数据。生成输出
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson' -
打印
hero_2。信息
即使
hero_2不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_2是新鲜的，它拥有所有可用数据。生成输出
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador' -
打印
hero_3。信息
即使
hero_3不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_3是新鲜的，它拥有所有可用数据。生成输出
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp' -
with块在这里结束（没有更多缩进代码），所以会话关闭，运行所有关闭代码。这包括执行任何可能已启动的事务的
ROLLBACK。生成输出
INFO Engine ROLLBACK -
打印行
"After the session closes"。生成输出
After the session closes -
在关闭会话后打印
hero_1。生成输出
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson' -
在关闭会话后打印
hero_2。生成输出
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador' -
在关闭会话后打印
hero_3。生成输出
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url, echo=True)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (1)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador") # (2)!
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48) # (3)!
print("Before interacting with the database") # (4)!
print("Hero 1:", hero_1) # (5)!
print("Hero 2:", hero_2) # (6)!
print("Hero 3:", hero_3) # (7)!
with Session(engine) as session: # (8)!
session.add(hero_1) # (9)!
session.add(hero_2) # (10)!
session.add(hero_3) # (11)!
print("After adding to the session") # (12)!
print("Hero 1:", hero_1) # (13)!
print("Hero 2:", hero_2) # (14)!
print("Hero 3:", hero_3) # (15)!
session.commit() # (16)!
print("After committing the session") # (17)!
print("Hero 1:", hero_1) # (18)!
print("Hero 2:", hero_2) # (19)!
print("Hero 3:", hero_3) # (20)!
print("After committing the session, show IDs") # (21)!
print("Hero 1 ID:", hero_1.id) # (22)!
print("Hero 2 ID:", hero_2.id) # (23)!
print("Hero 3 ID:", hero_3.id) # (24)!
print("After committing the session, show names") # (25)!
print("Hero 1 name:", hero_1.name) # (26)!
print("Hero 2 name:", hero_2.name) # (27)!
print("Hero 3 name:", hero_3.name) # (28)!
session.refresh(hero_1) # (29)!
session.refresh(hero_2) # (30)!
session.refresh(hero_3) # (31)!
print("After refreshing the heroes") # (32)!
print("Hero 1:", hero_1) # (33)!
print("Hero 2:", hero_2) # (34)!
print("Hero 3:", hero_3) # (35)!
# (36)!
print("After the session closes") # (37)!
print("Hero 1:", hero_1) # (38)!
print("Hero 2:", hero_2) # (39)!
print("Hero 3:", hero_3) # (40)!
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
-
创建
hero_1。不产生任何输出.
-
创建
hero_2。不产生任何输出.
-
创建
hero_3。不产生任何输出.
-
打印行
"Before interacting with the database"。生成输出
Before interacting with the database -
在与数据库交互之前打印
hero_1。生成输出
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None -
在与数据库交互之前打印
hero_2。生成输出
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None -
在与数据库交互之前打印
hero_3。生成输出
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48 -
在
with块中创建Session。不产生任何输出.
-
将
hero_1添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
将
hero_2添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
将
hero_3添加到会话。这仍然不会将其保存到数据库。
不产生任何输出.
-
打印行
"After adding to the session"。生成输出
After adding to the session -
在添加到会话后打印
hero_1。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None -
在添加到会话后打印
hero_2。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None -
在添加到会话后打印
hero_3。它仍然具有相同的数据，因为尚未与数据库进行任何交互。请注意，
id仍然是None。生成输出
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48 -
commit会话。这将保存所有数据到数据库。会话将使用引擎运行大量 SQL。
生成输出
INFO Engine BEGIN (implicit) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [generated in 0.00018s] ('Deadpond', 'Dive Wilson', None) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [cached since 0.0008968s ago] ('Spider-Boy', 'Pedro Parqueador', None) INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?) INFO Engine [cached since 0.001143s ago] ('Rusty-Man', 'Tommy Sharp', 48) INFO Engine COMMIT -
打印行
"After committing the session"。生成输出
After committing the session -
在提交会话后打印
hero_1。hero_1现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 1: -
在提交会话后打印
hero_2。hero_2现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 2: -
在提交会话后打印
hero_3。hero_3现在在内部被标记为已过期，在刷新之前，它看起来好像不包含任何数据。生成输出
Hero 3: -
打印行
"After committing the session, show IDs"。生成输出
After committing the session, show IDs -
打印
hero_1.id。这里发生了很多事情。因为我们正在访问
hero_1的属性id，SQLModel（实际上是 SQLAlchemy）可以检测到我们正在尝试访问hero_1的数据。然后它检测到
hero_1当前与一个会话关联（因为我们将其添加到会话并提交了它），并且它被标记为已过期。然后通过会话，它使用引擎执行所有 SQL，从数据库中获取此对象的数据。
接下来，它使用新数据更新对象，并在内部将其标记为“新鲜”或“未过期”。
最后，它使 ID 值可用于 Python 表达式的其余部分。在这种情况下，Python 表达式只是打印 ID。
生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00017s] (1,) Hero 1 ID: 1 -
打印
hero_2.id。这里发生了很多事情，所有与第 22 点相同的事情，但针对的是
hero_2对象。生成输出
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001245s ago] (2,) Hero 2 ID: 2 -
打印
hero_3.id。这里发生了很多事情，所有与第 22 点相同的事情，但针对的是
hero_3对象。生成输出
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.002215s ago] (3,) Hero 3 ID: 3 -
打印行
"After committing the session, show names"。生成输出
After committing the session, show names -
打印
hero_1.name。因为
hero_1仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 1 name: Deadpond -
打印
hero_2.name。因为
hero_2仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 2 name: Spider-Boy -
打印
hero_3.name。因为
hero_3仍然是新鲜的，所以没有获取额外的数据，没有执行额外的 SQL，并且名称可用。生成输出
Hero 3 name: Rusty-Man -
显式刷新
hero_1对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_1对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00024s] (1,) -
显式刷新
hero_2对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_2对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001487s ago] (2,) -
显式刷新
hero_3对象。会话将使用引擎执行必要的 SQL，从数据库中为
hero_3对象获取新鲜数据。生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.002377s ago] (3,) -
打印行
"After refreshing the heroes"。生成输出
After refreshing the heroes -
打印
hero_1。信息
即使
hero_1不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_1是新鲜的，它拥有所有可用数据。生成输出
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson' -
打印
hero_2。信息
即使
hero_2不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_2是新鲜的，它拥有所有可用数据。生成输出
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador' -
打印
hero_3。信息
即使
hero_3不新鲜，这也不会触发refresh，导致会话使用引擎从数据库中获取数据，因为它没有访问属性。因为
hero_3是新鲜的，它拥有所有可用数据。生成输出
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp' -
with块在这里结束（没有更多缩进代码），所以会话关闭，运行所有关闭代码。这包括执行任何可能已启动的事务的
ROLLBACK。生成输出
INFO Engine ROLLBACK -
打印行
"After the session closes"。生成输出
After the session closes -
在关闭会话后打印
hero_1。生成输出
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson' -
在关闭会话后打印
hero_2。生成输出
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador' -
在关闭会话后打印
hero_3。生成输出
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
这是运行此程序生成的所有输出，全部在一起
$ python app.py
INFO Engine BEGIN (implicit)
INFO Engine PRAGMA main.table_info("hero")
INFO Engine [raw sql] ()
INFO Engine PRAGMA temp.table_info("hero")
INFO Engine [raw sql] ()
INFO Engine
CREATE TABLE hero (
id INTEGER,
name VARCHAR NOT NULL,
secret_name VARCHAR NOT NULL,
age INTEGER,
PRIMARY KEY (id)
)
INFO Engine [no key 0.00018s] ()
INFO Engine COMMIT
Before interacting with the database
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48
After adding to the session
Hero 1: id=None name='Deadpond' secret_name='Dive Wilson' age=None
Hero 2: id=None name='Spider-Boy' secret_name='Pedro Parqueador' age=None
Hero 3: id=None name='Rusty-Man' secret_name='Tommy Sharp' age=48
INFO Engine BEGIN (implicit)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [generated in 0.00022s] ('Deadpond', 'Dive Wilson', None)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [cached since 0.001127s ago] ('Spider-Boy', 'Pedro Parqueador', None)
INFO Engine INSERT INTO hero (name, secret_name, age) VALUES (?, ?, ?)
INFO Engine [cached since 0.001483s ago] ('Rusty-Man', 'Tommy Sharp', 48)
INFO Engine COMMIT
After committing the session
Hero 1:
Hero 2:
Hero 3:
After committing the session, show IDs
INFO Engine BEGIN (implicit)
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [generated in 0.00029s] (1,)
Hero 1 ID: 1
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.002132s ago] (2,)
Hero 2 ID: 2
INFO Engine SELECT hero.id AS hero_id, hero.name AS hero_name, hero.secret_name AS hero_secret_name, hero.age AS hero_age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.003367s ago] (3,)
Hero 3 ID: 3
After committing the session, show names
Hero 1 name: Deadpond
Hero 2 name: Spider-Boy
Hero 3 name: Rusty-Man
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [generated in 0.00025s] (1,)
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.001583s ago] (2,)
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [cached since 0.002722s ago] (3,)
After refreshing the heroes
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson'
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador'
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
INFO Engine ROLLBACK
After the session closes
Hero 1: age=None id=1 name='Deadpond' secret_name='Dive Wilson'
Hero 2: age=None id=2 name='Spider-Boy' secret_name='Pedro Parqueador'
Hero 3: age=48 id=3 name='Rusty-Man' secret_name='Tommy Sharp'
回顾¶
你读完了所有这些！真是太多了！吃点蛋糕吧，这是你应得的。🍰
我们讨论了会话如何使用引擎向数据库发送 SQL，以创建数据和获取数据。它如何跟踪“已过期”和“新鲜”数据。它在哪些时刻自动获取数据（在访问实例属性时）以及数据如何通过会话在内存中的对象和数据库之间同步。
如果您理解了所有这些，那么您现在对 SQLModel、SQLAlchemy 以及 Python 与数据库交互的一般工作方式有了很多了解。
如果您没有完全理解，没关系，您可以随时回来 refresh 这些概念。
我认为这可能是导致问题和让您挠头的主要错误类型之一。所以，做得好，努力学习！💪