更新数据 - UPDATE¶
现在让我们看看如何使用 SQLModel 更新数据。
从之前的代码继续¶
和之前一样，我们将从之前代码的停止位置继续。
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.age <= 35)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.age <= 35)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
请记住在运行示例之前删除 database.db 文件，以获得相同的结果。
使用 SQL 更新¶
让我们快速检查一下如何使用 SQL 更新数据
UPDATE hero
SET age=16
WHERE name = "Spider-Boy"
这意味着，或多或少
嘿 SQL 数据库 👋，我想
UPDATE叫做hero的表。请
SETage列的值为16......对于每一行
WHERE列name的值等于"Spider-Boy"。
与 SELECT 语句类似，第一部分定义要处理的列：哪些列必须更新以及更新为哪个值。其余列保持不变。
第二部分，带有 WHERE，定义了应该将该更新应用于哪些行。
在这种情况下，由于我们只有一个名为 "Spider-Boy" 的 hero，因此它将仅在该行中应用更新。
信息
请注意，在 UPDATE 中，单等号 (=) 表示赋值，将列设置为某个值。
在 WHERE 中，相同的单等号 (=) 用于两个值之间的比较，以查找匹配的行。
这与 Python 和大多数编程语言形成对比，在 Python 和大多数编程语言中，单等号 (=) 用于赋值，而双等号 (==) 用于比较。
您可以在 DB Browser for SQLite 中尝试一下
更新之后，表中的数据将如下所示，Spider-Boy 的年龄为新年龄
| id | name | secret_name | age |
|---|---|---|---|
| 1 | Deadpond | Dive Wilson | null |
| 2 | Spider-Boy | Pedro Parqueador | 16 ✨ |
| 3 | Rusty-Man | Tommy Sharp | 48 |
提示
通过 id 查找要更新的行可能更常见，例如
UPDATE hero
SET age=16
WHERE id = 2
但是在上面的示例中，我使用了 name 以使其更直观。
现在让我们在代码中使用 SQLModel 进行相同的更新。
要获得相同的结果，请在运行示例之前删除 database.db 文件。
从数据库读取¶
我们将从选择 hero "Spider-Boy" 开始，这将是我们更新的对象
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
不要忘记将 update_heroes() 函数添加到 main() 函数中，以便我们在从命令行执行程序时调用它
# Code above omitted 👆
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
到那时，在命令行中运行它将输出
$ python app.py
// Some boilerplate and previous output omitted 😉
// The SELECT with WHERE
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.name = ?
INFO Engine [no key 0.00017s] ('Spider-Boy',)
// Print the hero as obtained from the database
Hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=None id=2
提示
请注意，到目前为止，hero 仍然没有年龄。
设置字段值¶
现在您有了一个 hero 对象，您可以简单地设置您想要的字段（表示列的属性）的值。
在这种情况下，我们将 age 设置为 16
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
将 Hero 添加到 Session¶
现在内存中的 hero 对象发生了更改，在本例中是 age 的新值，我们需要将其添加到 session。
这与我们创建新的 hero 实例时所做的相同
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
提交 Session¶
要保存在 session 中的当前更改，请提交它。
这将把更新后的 hero 保存在数据库中
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
它还将保存添加到 session 中的任何其他内容。
例如，如果您还创建了新的 hero，并且之前已将这些对象添加到 session 中，那么它们现在也会在这次提交中被保存。
此提交将生成此输出
$ python app.py
// Some boilerplate output omitted 😉
// Previous output omitted 🙈
// The SQL to update the hero in the database
INFO Engine UPDATE hero SET age=? WHERE hero.id = ?
INFO Engine [generated in 0.00017s] (16, 2)
INFO Engine COMMIT
刷新对象¶
此时，hero 已在数据库中更新，并且已在那里保存了新数据。
如果我们访问属性，例如 hero.name，对象中的数据将自动刷新。
但在此示例中，我们没有访问任何属性，我们只会打印对象。并且我们也想明确，因此我们将直接 .refresh() 对象
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
此刷新将触发与访问属性时自动触发的相同 SQL 查询。因此它将生成此输出
$ python app.py
// Some boilerplate output omitted 😉
// Previous output omitted 🙈
// The SQL to SELECT the fresh hero data
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
WHERE hero.id = ?
INFO Engine [generated in 0.00018s] (2,)
打印更新后的对象¶
现在我们可以直接打印 hero
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
# Code below omitted 👇
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
🤓 其他版本和变体
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
因为我们在更新后立即刷新了它，所以它具有新鲜数据，包括我们刚刚更新的新 age。
因此，打印它将显示新的 age
$ python app.py
// Some boilerplate output omitted 😉
// Previous output omitted 🙈
// Print the hero with the new age
Updated hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=16 id=2
代码回顾¶
现在让我们回顾一下所有代码
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero = results.one() # (3)!
print("Hero:", hero) # (4)!
hero.age = 16 # (5)!
session.add(hero) # (6)!
session.commit() # (7)!
session.refresh(hero) # (8)!
print("Updated hero:", hero) # (9)!
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
-
选择我们将要使用的 hero。
-
使用 select 语句对象执行查询。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00017s] ('Spider-Boy',) -
获取一个 hero 对象，期望只有一个。
提示
这确保不多于一个，并且只有一个，而不是
None。这永远不会返回
None，而是会引发异常。 -
打印 hero 对象。
这将生成输出
Hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=None id=2 -
将 hero 的 age 字段设置为新值
16。现在内存中的
hero对象对于 age 具有不同的值，但它仍未保存到数据库中。 -
将 hero 添加到 session。
这将其放在提交之前的 session 中的临时位置。
但它仍然没有保存在数据库中。
-
提交 session。
这将更新后的 hero 保存到数据库中。
这将生成输出
INFO Engine UPDATE hero SET age=? WHERE hero.id = ? INFO Engine [generated in 0.00017s] (16, 2) INFO Engine COMMIT -
刷新 hero 对象以获取最新数据，包括我们刚刚提交的 age。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00018s] (2,) -
打印更新后的 hero 对象。
这将生成输出
Updated hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=16 id=2
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero = results.one() # (3)!
print("Hero:", hero) # (4)!
hero.age = 16 # (5)!
session.add(hero) # (6)!
session.commit() # (7)!
session.refresh(hero) # (8)!
print("Updated hero:", hero) # (9)!
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
-
选择我们将要使用的 hero。
-
使用 select 语句对象执行查询。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00017s] ('Spider-Boy',) -
获取一个 hero 对象，期望只有一个。
提示
这确保不多于一个，并且只有一个，而不是
None。这永远不会返回
None，而是会引发异常。 -
打印 hero 对象。
这将生成输出
Hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=None id=2 -
将 hero 的 age 字段设置为新值
16。现在内存中的
hero对象对于 age 具有不同的值，但它仍未保存到数据库中。 -
将 hero 添加到 session。
这将其放在提交之前的 session 中的临时位置。
但它仍然没有保存在数据库中。
-
提交 session。
这将更新后的 hero 保存到数据库中。
这将生成输出
INFO Engine UPDATE hero SET age=? WHERE hero.id = ? INFO Engine [generated in 0.00017s] (16, 2) INFO Engine COMMIT -
刷新 hero 对象以获取最新数据，包括我们刚刚提交的 age。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00018s] (2,) -
打印更新后的 hero 对象。
这将生成输出
Updated hero: name='Spider-Boy' secret_name='Pedro Parqueador' age=16 id=2
提示
查看数字气泡以了解每行代码执行的操作。
多次更新¶
使用 SQLModel 进行更新过程与创建新对象的过程大致相同，您将它们添加到 session，然后提交它们。
这也意味着您可以一次更新多个字段（属性、列），并且还可以一次更新多个对象（hero）
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero_1 = results.one() # (3)!
print("Hero 1:", hero_1) # (4)!
statement = select(Hero).where(Hero.name == "Captain North America") # (5)!
results = session.exec(statement) # (6)!
hero_2 = results.one() # (7)!
print("Hero 2:", hero_2) # (8)!
hero_1.age = 16 # (9)!
hero_1.name = "Spider-Youngster" # (10)!
session.add(hero_1) # (11)!
hero_2.name = "Captain North America Except Canada" # (12)!
hero_2.age = 110 # (13)!
session.add(hero_2) # (14)!
session.commit() # (15)!
session.refresh(hero_1) # (16)!
session.refresh(hero_2) # (17)!
print("Updated hero 1:", hero_1) # (18)!
print("Updated hero 2:", hero_2) # (19)!
# (20)!
# Code below omitted 👇
-
选择 hero
Spider-Boy。 -
执行 select 语句。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00018s] ('Spider-Boy',) -
获取一个 hero 对象，对于 Spider-Boy 来说应该只有一个。
-
打印这个 hero。
这将生成输出
Hero 1: name='Spider-Boy' secret_name='Pedro Parqueador' age=None id=2 -
选择另一个 hero。
-
执行 select 语句。
这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00020s] ('Captain North America',)提示
看到顶部的
BEGIN了吗？这是 SQLAlchemy 自动为我们启动事务。
这样，即使创建它们的 SQL 已经发送到数据库，我们也可以在需要时回滚上次更改（如果有一些更改）。
-
为此新查询获取一个 hero 对象。
对于 Captain North America 来说应该只有一个。
-
打印第二个 hero。
这将生成输出
Hero 2: name='Captain North America' secret_name='Esteban Rogelios' age=93 id=7 -
更新第一个 hero 的年龄。
将属性
age的值设置为16。这会更新内存中的 hero 对象，但尚未更新数据库中的 hero 对象。
-
更新第一个 hero 的名称。
现在 hero 的名称将不再是
"Spider-Boy"，而是"Spider-Youngster"。同样，这会更新内存中的对象，但尚未更新数据库中的对象。
-
将第一个 hero 添加到 session。
这将其放在session 中的临时空间中，然后再将其提交到数据库。
它尚未保存。
-
更新第二个 hero 的名称。
现在 hero 的名称更加精确了。 😜
这会更新内存中的对象，但尚未更新数据库中的对象。
-
更新第二个 hero 的年龄。
这会更新内存中的对象，但尚未更新数据库中的对象。
-
将第二个 hero 添加到 session。
这将其放在session 中的临时空间中，然后再将其提交到数据库。
-
提交 session 中跟踪的所有更改。
这将一次性提交所有内容。
这将生成输出
INFO Engine UPDATE hero SET name=?, age=? WHERE hero.id = ? INFO Engine [generated in 0.00028s] (('Spider-Youngster', 16, 2), ('Captain North America Except Canada', 110, 7)) INFO Engine COMMIT提示
看看 SQLAlchemy（为 SQLModel 提供支持）如何优化 SQL 以在单个批处理中完成尽可能多的工作。
在这里，它在单个 SQL 查询中更新了两个 hero。
-
刷新第一个 hero。
这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00023s] (2,)提示
因为我们刚刚使用
COMMIT提交了 SQL 事务，所以 SQLAlchemy 将自动使用BEGIN启动新事务。 -
刷新第二个 hero。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001709s ago] (7,)提示
SQLAlchemy 仍在使用之前的事务，因此不必创建新事务。
-
打印第一个 hero，现在已更新。
这将生成输出
Updated hero 1: name='Spider-Youngster' secret_name='Pedro Parqueador' age=16 id=2 -
打印第二个 hero，现在已更新。
这将生成输出
Updated hero 2: name='Captain North America Except Canada' secret_name='Esteban Rogelios' age=110 id=7 -
这是
with块语句的结尾，因此 session 可以执行其终止代码。session 将
ROLLBACK（撤消）上次事务中任何可能的未提交更改。这将生成输出
INFO Engine ROLLBACK
# Code above omitted 👆
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero_1 = results.one() # (3)!
print("Hero 1:", hero_1) # (4)!
statement = select(Hero).where(Hero.name == "Captain North America") # (5)!
results = session.exec(statement) # (6)!
hero_2 = results.one() # (7)!
print("Hero 2:", hero_2) # (8)!
hero_1.age = 16 # (9)!
hero_1.name = "Spider-Youngster" # (10)!
session.add(hero_1) # (11)!
hero_2.name = "Captain North America Except Canada" # (12)!
hero_2.age = 110 # (13)!
session.add(hero_2) # (14)!
session.commit() # (15)!
session.refresh(hero_1) # (16)!
session.refresh(hero_2) # (17)!
print("Updated hero 1:", hero_1) # (18)!
print("Updated hero 2:", hero_2) # (19)!
# (20)!
# Code below omitted 👇
-
选择 hero
Spider-Boy。 -
执行 select 语句。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00018s] ('Spider-Boy',) -
获取一个 hero 对象，对于 Spider-Boy 来说应该只有一个。
-
打印这个 hero。
这将生成输出
Hero 1: name='Spider-Boy' secret_name='Pedro Parqueador' age=None id=2 -
选择另一个 hero。
-
执行 select 语句。
这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.name = ? INFO Engine [no key 0.00020s] ('Captain North America',)提示
看到顶部的
BEGIN了吗？这是 SQLAlchemy 自动为我们启动事务。
这样，即使创建它们的 SQL 已经发送到数据库，我们也可以在需要时回滚上次更改（如果有一些更改）。
-
为此新查询获取一个 hero 对象。
对于 Captain North America 来说应该只有一个。
-
打印第二个 hero。
这将生成输出
Hero 2: name='Captain North America' secret_name='Esteban Rogelios' age=93 id=7 -
更新第一个 hero 的年龄。
将属性
age的值设置为16。这会更新内存中的 hero 对象，但尚未更新数据库中的 hero 对象。
-
更新第一个 hero 的名称。
现在 hero 的名称将不再是
"Spider-Boy"，而是"Spider-Youngster"。同样，这会更新内存中的对象，但尚未更新数据库中的对象。
-
将第一个 hero 添加到 session。
这将其放在session 中的临时空间中，然后再将其提交到数据库。
它尚未保存。
-
更新第二个 hero 的名称。
现在 hero 的名称更加精确了。 😜
这会更新内存中的对象，但尚未更新数据库中的对象。
-
更新第二个 hero 的年龄。
这会更新内存中的对象，但尚未更新数据库中的对象。
-
将第二个 hero 添加到 session。
这将其放在session 中的临时空间中，然后再将其提交到数据库。
-
提交 session 中跟踪的所有更改。
这将一次性提交所有内容。
这将生成输出
INFO Engine UPDATE hero SET name=?, age=? WHERE hero.id = ? INFO Engine [generated in 0.00028s] (('Spider-Youngster', 16, 2), ('Captain North America Except Canada', 110, 7)) INFO Engine COMMIT提示
看看 SQLAlchemy（为 SQLModel 提供支持）如何优化 SQL 以在单个批处理中完成尽可能多的工作。
在这里，它在单个 SQL 查询中更新了两个 hero。
-
刷新第一个 hero。
这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [generated in 0.00023s] (2,)提示
因为我们刚刚使用
COMMIT提交了 SQL 事务，所以 SQLAlchemy 将自动使用BEGIN启动新事务。 -
刷新第二个 hero。
这将生成输出
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero WHERE hero.id = ? INFO Engine [cached since 0.001709s ago] (7,)提示
SQLAlchemy 仍在使用之前的事务，因此不必创建新事务。
-
打印第一个 hero，现在已更新。
这将生成输出
Updated hero 1: name='Spider-Youngster' secret_name='Pedro Parqueador' age=16 id=2 -
打印第二个 hero，现在已更新。
这将生成输出
Updated hero 2: name='Captain North America Except Canada' secret_name='Esteban Rogelios' age=110 id=7 -
这是
with块语句的结尾，因此 session 可以执行其终止代码。session 将
ROLLBACK（撤消）上次事务中任何可能的未提交更改。这将生成输出
INFO Engine ROLLBACK
👀 完整文件预览
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero_1 = results.one() # (3)!
print("Hero 1:", hero_1) # (4)!
statement = select(Hero).where(Hero.name == "Captain North America") # (5)!
results = session.exec(statement) # (6)!
hero_2 = results.one() # (7)!
print("Hero 2:", hero_2) # (8)!
hero_1.age = 16 # (9)!
hero_1.name = "Spider-Youngster" # (10)!
session.add(hero_1) # (11)!
hero_2.name = "Captain North America Except Canada" # (12)!
hero_2.age = 110 # (13)!
session.add(hero_2) # (14)!
session.commit() # (15)!
session.refresh(hero_1) # (16)!
session.refresh(hero_2) # (17)!
print("Updated hero 1:", hero_1) # (18)!
print("Updated hero 2:", hero_2) # (19)!
# (20)!
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
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)
hero_4 = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_5 = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_6 = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_7 = Hero(name="Captain North America", secret_name="Esteban Rogelios", age=93)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.add(hero_4)
session.add(hero_5)
session.add(hero_6)
session.add(hero_7)
session.commit()
def update_heroes():
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy") # (1)!
results = session.exec(statement) # (2)!
hero_1 = results.one() # (3)!
print("Hero 1:", hero_1) # (4)!
statement = select(Hero).where(Hero.name == "Captain North America") # (5)!
results = session.exec(statement) # (6)!
hero_2 = results.one() # (7)!
print("Hero 2:", hero_2) # (8)!
hero_1.age = 16 # (9)!
hero_1.name = "Spider-Youngster" # (10)!
session.add(hero_1) # (11)!
hero_2.name = "Captain North America Except Canada" # (12)!
hero_2.age = 110 # (13)!
session.add(hero_2) # (14)!
session.commit() # (15)!
session.refresh(hero_1) # (16)!
session.refresh(hero_2) # (17)!
print("Updated hero 1:", hero_1) # (18)!
print("Updated hero 2:", hero_2) # (19)!
# (20)!
def main():
create_db_and_tables()
create_heroes()
update_heroes()
if __name__ == "__main__":
main()
提示
通过单击代码中的每个数字气泡来查看每行代码的作用。 👆
回顾¶
更新 SQLModel 对象就像您处理其他 Python 对象一样。 🐍
只需记住将它们 add 到 session，然后 commit 它。如果需要，refresh 它们。