读取数据 - SELECT¶
我们已经有一个数据库和一个表，其中包含了一些数据，大致如下所示
| id | name | secret_name | age |
|---|---|---|---|
| 1 | 死侍 | 戴夫·威尔逊 | 空 |
| 2 | 蜘蛛男孩 | 佩德罗·帕尔克多 | 空 |
| 3 | 锈人 | 汤米·夏普 | 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"的表中FROM获取它们。
然后数据库将获取数据并以如下表格形式返回给你
| id | name | secret_name | age |
|---|---|---|---|
| 1 | 死侍 | 戴夫·威尔逊 | 空 |
| 2 | 蜘蛛男孩 | 佩德罗·帕尔克多 | 空 |
| 3 | 锈人 | 汤米·夏普 | 48 |
你可以在 DB Browser for SQLite 中尝试一下
警告
这里我们获取了所有行。
如果你有数千行，这对于数据库来说计算成本可能很高。
你通常会希望过滤行，只接收你想要的那些。但我们将在下一章中学习这一点。
SQL 快捷方式¶
如果我们想要像上面那样获取所有列，在 SQL 中有一个快捷方式，不是指定每个列名，我们可以写一个 *
SELECT *
FROM hero
那会得到相同的结果。虽然我们不会将它用于 SQLModel。
SELECT 更少的列¶
我们还可以 SELECT 更少的列，例如
SELECT id, name
FROM hero
这里我们只选择了 id 和 name 列。
它会生成一个如下所示的表格
| id | name |
|---|---|
| 1 | 死侍 |
| 2 | 蜘蛛男孩 |
| 3 | 锈人 |
这里有一个有趣的地方值得注意。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。
另一个变体是，大多数 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 表中选择 FROM 已经很明显了，因为这是与 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 | 死侍 | 戴夫·威尔逊 | 空 |
| 2 | 蜘蛛男孩 | 佩德罗·帕尔克多 | 空 |
| 3 | 锈人 | 汤米·夏普 | 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)。我们整个应用程序仍然只有一个引擎。
-
使用
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)。我们整个应用程序仍然只有一个引擎。
-
使用
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)，这里实际上没有必要有两个。
但这让我可以向你展示它们是如何分离的，并强化你每个应用程序应该有一个引擎 (engine)，以及多个会话 (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 结合使用（使用相同的模型等）。