读取数据 - SELECT¶
我们已经有了一个数据库和一个包含一些数据的表，它看起来或多或少像这样
| id | name | secret_name | age |
|---|---|---|---|
| 1 | Deadpond | Dive Wilson | null |
| 2 | Spider-Boy | Pedro Parqueador | null |
| 3 | Rusty-Man | Tommy Sharp | 48 |
事情变得越来越令人兴奋了！现在让我们看看如何从数据库中读取数据！🤩
从之前的代码继续¶
让我们从上次用来创建一些数据的代码继续。
👀 完整文件预览
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def main():
create_db_and_tables()
create_heroes()
if __name__ == "__main__":
main()
我们正在创建一个 SQLModel Hero 类模型并创建一些记录。
我们将需要 Hero 模型和 engine，但我们将创建一个新的会话来在新函数中查询数据。
使用 SQL 读取数据¶
在编写 Python 代码之前，让我们快速回顾一下使用 SQL 查询数据的方式
SELECT id, name, secret_name, age
FROM hero
它的意思大致是
嘿 SQL 数据库 👋，请去帮我
SELECT一些数据。我将首先告诉你我想要的列
idnamesecret_nameage我希望你从名为
"hero"的表中获取它们。
然后数据库将去获取数据，并以如下表格形式返回给你
| id | name | secret_name | age |
|---|---|---|---|
| 1 | Deadpond | Dive Wilson | null |
| 2 | Spider-Boy | Pedro Parqueador | null |
| 3 | Rusty-Man | Tommy Sharp | 48 |
你可以在 DB Browser for SQLite 中尝试一下
警告
这里我们获取了所有行。
如果你有成千上万行，那么对于数据库来说，计算成本可能会很高。
你通常会希望过滤行，只接收你想要的行。但我们将在下一章中学习这一点。
SQL 快捷方式¶
如果我们想要像上面这种情况一样获取所有列，在 SQL 中有一个快捷方式，我们可以写一个 * 来代替指定每个列名
SELECT *
FROM hero
这将得到相同的结果。虽然我们不会将它用于 SQLModel。
SELECT 更少的列¶
我们也可以 SELECT 更少的列，例如
SELECT id, name
FROM hero
这里我们只选择了 id 和 name 列。
它将产生如下表格
| id | name |
|---|---|
| 1 | Deadpond |
| 2 | Spider-Boy |
| 3 | Rusty-Man |
这里有一些有趣的地方需要注意。SQL 数据库将其数据存储在表中。它们也总是以表格形式传达其结果。
SELECT 变体¶
SQL 语言在多个地方允许一些变体。
其中一个变体是在 SELECT 语句中，你可以直接使用列名，也可以在列名前面加上表名和一个点。
例如，上面的相同 SQL 代码可以写成
SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
当同时处理可能具有相同列名（例如 hero.id 和 team.id，或者 hero.name 和 team.name）的多个表时，这一点尤其重要。
例如 hero.id 和 team.id，或者 hero.name 和 team.name。
另一个变体是，大多数 SQL 关键字（如 SELECT）也可以用小写字母书写，如 select。
结果表不必存在¶
这是有趣的部分。SQL 数据库返回的表不必作为独立表存在于数据库中。🧙
例如，在我们的数据库中，我们只有一个表，它包含所有列：id、name、secret_name、age。在这里，我们得到一个列数较少的结果表。
SQL 的主要目的之一是能够将数据结构化地保存在不同的表中，而无需重复数据等，然后以多种方式查询数据库，并获得许多不同的表格作为结果。
使用 SQLModel 读取数据¶
现在让我们做同样的查询来读取所有英雄，但是使用 SQLModel。
创建 Session¶
第一步是创建 Session，就像我们在创建行时所做的那样。
我们将从一个新的函数 select_heroes() 开始
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
创建 select 语句¶
接下来，几乎与我们上面编写 SQL SELECT 语句的方式相同，现在我们将创建一个 SQLModel select 语句。
首先，我们必须从文件顶部的 sqlmodel 导入 select
from sqlmodel import Field, Session, SQLModel, create_engine, select
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
然后我们将使用它在 Python 代码中创建一个 SELECT 语句
from sqlmodel import Field, Session, SQLModel, create_engine, select
# Code here omitted 👈
def select_heroes():
with Session(engine) as session:
statement = select(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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
这是一个非常简单的代码行，但传达了很多信息
statement = select(Hero)
这等效于上面的第一个 SQL SELECT 语句
SELECT id, name, secret_name, age
FROM hero
我们将类模型 Hero 传递给 select() 函数。这告诉它我们想要选择 Hero 类所需的所有列。
请注意，在 select() 函数中，我们没有显式指定 FROM 部分。对于 SQLModel（实际上是 SQLAlchemy）来说，我们想要从表 hero 中选择数据已经很明显了，因为这是与 Hero 类模型关联的表。
提示
select() 返回的 statement 值是一个特殊对象，允许我们执行其他操作。
我将在接下来的章节中告诉你这一点。
执行语句¶
现在我们有了 select 语句，我们可以使用 session 执行它
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
这将告诉 session 继续并使用 engine 在数据库中执行该 SELECT 语句并将结果带回。
因为我们使用 echo=True 创建了 engine，所以它将在输出中显示它执行的 SQL。
这个 session.exec(statement) 将生成以下输出
INFO Engine BEGIN (implicit)
INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age
FROM hero
INFO Engine [no key 0.00032s] ()
数据库返回包含所有数据的表，就像我们上面直接编写 SQL 时一样
| id | name | secret_name | age |
|---|---|---|---|
| 1 | Deadpond | Dive Wilson | null |
| 2 | Spider-Boy | Pedro Parqueador | null |
| 3 | Rusty-Man | Tommy Sharp | 48 |
遍历结果¶
results 对象是一个 可迭代对象，可以用来遍历每一行。
现在我们可以把它放在 for 循环中，并打印每个英雄
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
这将打印输出
id=1 name='Deadpond' age=None secret_name='Dive Wilson'
id=2 name='Spider-Boy' age=None secret_name='Pedro Parqueador'
id=3 name='Rusty-Man' age=48 secret_name='Tommy Sharp'
将 select_heroes() 添加到 main()¶
现在在 main() 函数中包含对 select_heroes() 的调用，以便在我们从命令行运行程序时执行它
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
for hero in results:
print(hero)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
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 sqlmodel import Field, Session, SQLModel, create_engine, select # (1)!
class Hero(SQLModel, table=True): # (2)!
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) # (3)!
def create_db_and_tables():
SQLModel.metadata.create_all(engine) # (4)!
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (5)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
with Session(engine) as session: # (6)!
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session: # (7)!
statement = select(Hero) # (8)!
results = session.exec(statement) # (9)!
for hero in results: # (10)!
print(hero) # (11)!
# (12)!
def main():
create_db_and_tables()
create_heroes()
select_heroes() # (13)!
if __name__ == "__main__":
main()
-
从
sqlmodel导入我们将要使用的所有内容，包括新的select()函数。 -
创建
Hero类模型，表示hero表。 -
创建 engine，我们应该使用一个由所有应用程序代码共享的引擎，而这正是我们在这里所做的。
-
为
SQLModel.metadata中注册的模型创建所有表。如果数据库尚不存在，这也会创建数据库。
-
创建每个
Hero对象。如果你已经在数据库中创建了数据，你的版本中可能没有这个。
-
创建一个新的 session 并使用它将英雄
add到数据库，然后commit更改。 -
创建一个新的 session 来查询数据。
提示
请注意，这是一个新的 session，独立于上面另一个函数中的会话。
但它仍然使用相同的 engine。我们仍然为整个应用程序使用一个 engine。
-
使用
select()函数创建一个语句，选择所有Hero对象。这将选择
hero表中的所有行。 -
使用
session.exec(statement)使 session 使用 engine 来执行内部 SQL 语句。这将进入数据库，执行该 SQL，并将结果返回。
它返回一个特殊的迭代对象，我们将其放入变量
results中。这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero INFO Engine [no key 0.00032s] () -
为
results中的每个Hero对象迭代。 -
打印每个
hero。for循环中的 3 次迭代将生成以下输出id=1 name='Deadpond' age=None secret_name='Dive Wilson' id=2 name='Spider-Boy' age=None secret_name='Pedro Parqueador' id=3 name='Rusty-Man' age=48 secret_name='Tommy Sharp' -
此时，在
with块之后，session 已关闭。这将生成输出
INFO Engine ROLLBACK -
将此函数
select_heroes()添加到main()函数，以便在我们从命令行运行此程序时调用它。
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select # (1)!
class Hero(SQLModel, table=True): # (2)!
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) # (3)!
def create_db_and_tables():
SQLModel.metadata.create_all(engine) # (4)!
def create_heroes():
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson") # (5)!
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
with Session(engine) as session: # (6)!
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session: # (7)!
statement = select(Hero) # (8)!
results = session.exec(statement) # (9)!
for hero in results: # (10)!
print(hero) # (11)!
# (12)!
def main():
create_db_and_tables()
create_heroes()
select_heroes() # (13)!
if __name__ == "__main__":
main()
-
从
sqlmodel导入我们将要使用的所有内容，包括新的select()函数。 -
创建
Hero类模型，表示hero表。 -
创建 engine，我们应该使用一个由所有应用程序代码共享的引擎，而这正是我们在这里所做的。
-
为
SQLModel.metadata中注册的模型创建所有表。如果数据库尚不存在，这也会创建数据库。
-
创建每个
Hero对象。如果你已经在数据库中创建了数据，你的版本中可能没有这个。
-
创建一个新的 session 并使用它将英雄
add到数据库，然后commit更改。 -
创建一个新的 session 来查询数据。
提示
请注意，这是一个新的 session，独立于上面另一个函数中的会话。
但它仍然使用相同的 engine。我们仍然为整个应用程序使用一个 engine。
-
使用
select()函数创建一个语句，选择所有Hero对象。这将选择
hero表中的所有行。 -
使用
session.exec(statement)使 session 使用 engine 来执行内部 SQL 语句。这将进入数据库，执行该 SQL，并将结果返回。
它返回一个特殊的迭代对象，我们将其放入变量
results中。这将生成输出
INFO Engine BEGIN (implicit) INFO Engine SELECT hero.id, hero.name, hero.secret_name, hero.age FROM hero INFO Engine [no key 0.00032s] () -
为
results中的每个Hero对象迭代。 -
打印每个
hero。for循环中的 3 次迭代将生成以下输出id=1 name='Deadpond' age=None secret_name='Dive Wilson' id=2 name='Spider-Boy' age=None secret_name='Pedro Parqueador' id=3 name='Rusty-Man' age=48 secret_name='Tommy Sharp' -
此时，在
with块之后，session 已关闭。这将生成输出
INFO Engine ROLLBACK -
将此函数
select_heroes()添加到main()函数，以便在我们从命令行运行此程序时调用它。
提示
查看数字气泡以了解每行代码的作用。
在这里，为什么我们应该为整个应用程序使用一个 engine，但为每组操作使用不同的 session 开始变得更加明显。
我们创建的这个新会话使用相同的 engine，但它是一个新的且独立的 session。
例如，上面创建模型的代码可以存在于处理 Web API 请求和创建模型的函数中。
而从数据库读取数据的第二部分可能位于另一个函数中，用于其他请求。
因此，这两个部分可能位于不同的位置，并且需要它们自己的会话。
信息
公平地说，在这个例子中，所有这些代码实际上可以共享同一个 session，实际上这里没有必要有两个 session。
但这让我可以向你展示它们是如何分离的，并加强你应该为每个应用程序使用一个 engine，以及多个 session，每个操作组一个 session 的想法。
获取 Hero 对象列表¶
到目前为止，我们正在使用 results 来迭代它们。
但是出于不同的原因，你可能希望立即拥有完整的 Hero 对象列表，而不仅仅是可迭代对象。例如，如果你想在 Web API 中返回它们。
特殊的 results 对象还有一个方法 results.all()，它返回一个包含所有对象的列表
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
heroes = results.all()
print(heroes)
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
heroes = results.all()
print(heroes)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
statement = select(Hero)
results = session.exec(statement)
heroes = results.all()
print(heroes)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
有了这个，我们现在在 heroes 变量中拥有所有英雄的列表。
打印后，我们将看到类似这样的内容
[
Hero(id=1, name='Deadpond', age=None, secret_name='Dive Wilson'),
Hero(id=2, name='Spider-Boy', age=None, secret_name='Pedro Parqueador'),
Hero(id=3, name='Rusty-Man', age=48, secret_name='Tommy Sharp')
]
信息
它实际上看起来更紧凑，我稍微格式化了一下，以便你可以看到它实际上是一个包含所有数据的列表。
紧凑版本¶
我创建了几个变量，以便能够向你解释每件事都在做什么。
但是了解每个对象是什么以及它都在做什么，我们可以稍微简化一下，并将其放入更紧凑的形式中
# Code above omitted 👆
def select_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
print(heroes)
# 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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
print(heroes)
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
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)
with Session(engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
def select_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
print(heroes)
def main():
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
这里我们将其全部放在一行中，你可能会更频繁地将 select 语句放在像这样的一行中。
SQLModel 或 SQLAlchemy - 技术细节¶
SQLModel 实际上或多或少只是 SQLAlchemy 和 Pydantic 的底层组合。
它使用并返回相同类型的对象，并且与这两个库兼容。
尽管如此，SQLModel 定义了一些自己的内部部件，以改善开发者体验。
在本章中，我们将接触到其中的一些。
SQLModel 的 select¶
当从 sqlmodel 导入 select() 函数时，你正在使用 SQLModel 版本的 select。
SQLAchemy 也有自己的 select，SQLModel 的 select 在内部使用 SQLAlchemy 的 select。
但是 SQLModel 的版本使用类型注解做了很多技巧，以确保你获得尽可能最佳的编辑器支持，无论你使用 VS Code、PyCharm 还是其他工具。✨
信息
为了尽可能地改进这一点，我们进行了大量的工作和研究，使用了不同版本的内部代码。🤓
SQLModel 的 session.exec¶
📢 这是一个需要特别注意的地方。
SQLAlchemy 自己的 Session 有一个方法 session.execute()。它没有 session.exec() 方法。
如果你看到 SQLAlchemy 教程，他们总是会使用 session.execute()。
SQLModel 自己的 Session 直接继承自 SQLAlchemy 的 Session，并添加了这个额外的方法 session.exec()。在底层，它使用相同的 session.execute()。
但是 session.exec() 结合 session() 中的技巧，做了一些技巧，为你提供最佳的编辑器支持，在任何地方都具有自动完成和内联错误，即使在从 select 获取数据之后也是如此。✨
例如，在 SQLAlchemy 中，你需要在这里添加一个 .scalars()
heroes = session.execute(select(Hero)).scalars().all()
但是当选择多个事物时（我们稍后会看到），你必须删除它。
SQLModel 的 session.exec() 为你处理了这个问题，所以你无需添加 .scalars()。
这是 SQLAlchemy 目前无法提供的功能，因为常规的 session.execute() 支持其他几个用例，包括遗留用例，因此它不能拥有所有的内部类型注解和技巧来支持这一点。
最重要的是，SQLModel 的 session.exec() 也做了一些技巧，以减少你必须编写的代码量，并使其尽可能直观。
但是 SQLModel 的 Session 仍然可以访问 session.execute()。
提示
你的编辑器将为你提供 session.exec() 和 session.execute() 的自动完成。
📢 记住始终使用 session.exec() 以获得最佳的编辑器支持和开发者体验。
SQLModel 风格的注意事项¶
SQLModel 旨在在非常常见的用例的狭窄范围内提供最佳的开发者体验。✨
当你需要时，你仍然可以将其与 SQLAlchemy 直接结合使用，并使用 SQLAlchemy 的所有功能，包括更底层的“纯”SQL 构造、异构模式，甚至遗留模式。🤓
但是 SQLModel 的设计（例如类型注解）假设你正在以我在文档中解释的方式使用它。
因此，你将获得尽可能多的自动完成和内联错误。🚀
但这也意味着，如果你将 SQLModel 与 SQLAlchemy 的一些更异构的模式一起使用，你的编辑器可能会告诉你存在错误，而事实上，代码仍然可以工作。
这就是权衡。🤷
但是对于你需要那些异构模式的情况，你始终可以将 SQLAlchemy 直接与 SQLModel 结合使用（使用相同的模型等）。