设置¶

初始化应用¶

定义所有模型后，tortoise 需要你初始化它们，以便在模型之间创建反向关系，并将你的数据库客户端与相应的模型匹配。

你可以像这样操作

from tortoise import Tortoise

async def init():
    # Here we create a SQLite DB using file "db.sqlite3"
    #  also specify the app name of "models"
    #  which contain models from "app.models"
    await Tortoise.init(
        db_url='sqlite://db.sqlite3',
        modules={'models': ['app.models']}
    )
    # Generate the schema
    await Tortoise.generate_schemas()

这里我们创建到 SQLite 数据库客户端的连接，然后我们发现并初始化模型。

generate_schema 在空数据库中生成模式，你不应该在每次应用初始化时运行它，只运行一次，可能在你的主代码之外。在生成模式时，还可以将 safe 参数设置为 True，这将仅在表不存在时插入表。

如果你在 app.models 模块（或你指定从中加载模型的任何位置）中定义变量 __models__，generate_schema 将使用该列表，而不是自动为你查找模型。

清理的重要性¶

Tortoise ORM 将保持与外部数据库的连接处于打开状态。作为 asyncio Python 库，它需要正确关闭连接，否则 Python 解释器可能仍在等待所述连接完成。

为确保关闭连接，请确保调用 Tortoise.close_connections()

await Tortoise.close_connections()

小助手函数 tortoise.run_async() 将确保关闭连接。

参考¶

class tortoise.BaseDBAsyncClient(connection_name, fetch_inserted=True, **kwargs)[source]¶

用于包含数据库连接的基本类。

参数作为 kwargs 传递，并且主要是特定于驱动程序的。

query_class Type[pypika.Query]¶

PyPika 查询方言（低级方言）

executor_class Type[BaseExecutor]¶

执行器方言类（高级方言）

schema_generator Type[BaseSchemaGenerator]¶

DDL 架构生成器

capabilities Capabilities¶

包含连接功能

acquire_connection()[source]¶

从池中获取连接。如果已在事务中，将返回当前上下文连接。

返回类型:¶

Union[ConnectionWrapper, PoolConnectionWrapper]

capabilities : Capabilities = <tortoise.backends.base.client.Capabilities object>¶

async close()[source]¶

关闭数据库连接。

返回类型:¶

无

async create_connection(with_db)[source]¶

建立数据库连接。

参数:¶

with_db¶

如果为 True，则选择要使用的数据库，否则使用默认值。此用例用于创建/删除数据库。

返回类型:¶

无

async db_create()[source]¶

在服务器中创建数据库。通常仅由测试运行器调用。

需要使用参数 with_db=False 调用 create_connection()` 以使用默认连接而不是配置的连接，否则你会收到指示数据库不存在的错误。

返回类型:¶

无

async db_delete()[source]¶

从服务器中删除数据库。通常仅由测试运行器调用。

需要使用参数 with_db=False 调用 create_connection()` 以使用默认连接而不是配置的连接，否则你会收到指示数据库正在使用的错误。

返回类型:¶

无

async execute_insert(query, values)[source]¶

执行 RAW SQL 插入语句，并提供参数。

参数:¶

query¶

针对目标数据库方言预参数化的 SQL 字符串。

values¶

位置 DB 参数序列。

返回类型:¶

Any

返回:¶

如果是 DB 生成的主键。（目前仅为整数自动编号 PK）

async execute_many(query, values)[source]¶

执行 RAW 批量插入语句，类似于 execute_insert，但不返回数据。

参数:¶

query¶

针对目标数据库方言预参数化的 SQL 字符串。

values¶

位置 DB 参数序列。

返回类型:¶

无

async execute_query(query, values=None)[source]¶

执行 RAW SQL 查询语句，并返回结果集。

参数:¶

query¶

针对目标数据库方言预参数化的 SQL 字符串。

values=None¶

位置 DB 参数序列。

返回类型:¶

Tuple[int, Sequence[dict]]

返回:¶

元组：(受影响的行数，结果集)

async execute_query_dict(query, values=None)[source]¶

执行 RAW SQL 查询语句，并将结果集作为字典列表返回。

参数:¶

query¶

针对目标数据库方言预参数化的 SQL 字符串。

values=None¶

位置 DB 参数序列。

返回类型:¶

List[dict]

async execute_script(query)[source]¶

执行包含多个语句的 RAW SQL 脚本，不返回任何内容。

参数:¶

query¶

将原样传递的 SQL 字符串。此处支持分号。

返回类型:¶

无

executor_class¶

BaseExecutor 的别名

query_class¶

Query 的别名

schema_generator¶

别名 BaseSchemaGenerator

class tortoise.Model(**kwargs)[source]¶

所有 Tortoise ORM 模型的基本类。

class Meta[source]¶

Meta 类用于配置模型的元数据。

用法

class Foo(Model):
    ...

    class Meta:
        table="custom_table"
        unique_together=(("field_a", "field_b"), )

classmethod all(using_db=None)[source]¶

返回完整 QuerySet。

返回类型:¶

QuerySet[typing_extensions.Self]

classmethod annotate(**kwargs)[source]¶

使用额外的函数/聚合/表达式对结果集进行注释。

参数:¶

**kwargs¶

参数名称和要注释的函数/聚合。

返回类型:¶

QuerySet[typing_extensions.Self]

类方法 bulk_create(objects, batch_size=None, ignore_conflicts=False, update_fields=None, on_conflict=None, using_db=None)[源代码]¶

批量插入操作

注意

批量插入操作将执行最少操作以确保在数据库中创建的对象具有所有默认值和已设置的生成字段，但在 Python 中可能是不完整的引用。

例如 IntField 主键将不会填充。

仅建议在您希望确保最佳插入性能的一次性插入中使用此功能。

User.bulk_create([
    User(name="...", email="..."),
    User(name="...", email="...")
])

参数:¶

on_conflict=None¶

冲突索引名称

update_fields=None¶

冲突时更新字段

ignore_conflicts=False¶

插入时忽略冲突

objects¶

要批量创建的对象列表

batch_size=None¶

单个查询中创建的对象数量

using_db=None¶

特定数据库连接，用于替代默认绑定

返回类型:¶

BulkCreateQuery[Model]

classmethod bulk_update(objects, fields, batch_size=None, using_db=None)[源代码]¶

更新数据库中每个给定对象中的给定字段。此方法有效地更新提供模型实例上的给定字段，通常使用一个查询。

users = [
    await User.create(name="...", email="..."),
    await User.create(name="...", email="...")
]
users[0].name = 'name1'
users[1].name = 'name2'

await User.bulk_update(users, fields=['name'])

参数:¶

objects¶

要批量创建的对象列表

fields¶

要更新的字段

batch_size=None¶

单个查询中创建的对象数量

using_db=None¶

特定数据库连接，用于替代默认绑定

返回类型:¶

BulkUpdateQuery[Model]

clone(pk=<object object>)[source]¶

创建对象的全新克隆，当您执行 .save() 时，将创建一个新记录。

参数:¶

pk : Any

如果模型不生成自己的主键，则为可选必需值。此处指定的任何值都将始终被使用。

返回类型:¶

Model

返回:¶

不包含主键信息的当前对象副本。

引发:¶

ParamsError – 如果需要 pk 但未提供。

async classmethod create(using_db=None, **kwargs)[source]¶

在 DB 中创建一条记录并返回对象。

user = await User.create(name="...", email="...")

等效于

user = User(name="...", email="...")
await user.save()

参数:¶

using_db=None¶

特定数据库连接，用于替代默认绑定

**kwargs¶

模型参数。

返回类型:¶

Model

async delete(using_db=None)[source]¶

删除当前模型对象。

参数:¶

using_db=None¶

特定数据库连接，用于替代默认绑定

引发:¶

OperationalError – 如果对象从未持久化。

返回类型:¶

无

classmethod describe(serializable=True)[source]¶

描述给定的模型列表或所有已注册的模型。

参数:¶

serializable=True¶

False 如果想要原始 Python 对象，True 如果想要可 JSON 序列化的数据。（默认为 True）

返回类型:¶

dict

返回:¶

包含模型描述的字典。

基本字典有一组固定的键，它们引用字段列表（或在主键的情况下引用单个字段）

{
    "name":                 str     # Qualified model name
    "app":                  str     # 'App' namespace
    "table":                str     # DB table name
    "abstract":             bool    # Is the model Abstract?
    "description":          str     # Description of table (nullable)
    "docstring":            str     # Model docstring (nullable)
    "unique_together":      [...]   # List of List containing field names that
                                    #  are unique together
    "pk_field":             {...}   # Primary key field
    "data_fields":          [...]   # Data fields
    "fk_fields":            [...]   # Foreign Key fields FROM this model
    "backward_fk_fields":   [...]   # Foreign Key fields TO this model
    "o2o_fields":           [...]   # OneToOne fields FROM this model
    "backward_o2o_fields":  [...]   # OneToOne fields TO this model
    "m2m_fields":           [...]   # Many-to-Many fields
}

每个字段的指定方式如 tortoise.fields.base.Field.describe() 中所定义

classmethod exclude(*args, **kwargs)[source]¶

生成一个应用了 exclude 的 QuerySet。

参数:¶

*args¶

包含约束的 Q 函数。将进行 AND 操作。

**kwargs¶

简单筛选约束。

返回类型:¶

QuerySet[typing_extensions.Self]

classmethod exists(*args, using_db=None, **kwargs)[source]¶

返回 True/False，表示使用提供的筛选参数是否存在记录。

result = await User.exists(username="foo")

参数:¶

using_db=None¶

要使用的特定数据库连接。

*args¶

包含约束的 Q 函数。将进行 AND 操作。

**kwargs¶

简单筛选约束。

返回类型:¶

ExistsQuery

async classmethod fetch_for_list(instance_list, *args, using_db=None)[源代码]¶

为提供的 Model 对象列表获取相关模型。

参数:¶

instance_list¶

要为其获取关系的 Model 对象列表。

*args¶

要获取的关系名称。

using_db=None¶

不要使用

返回类型:¶

无

async fetch_related(*args, using_db=None)[源代码]¶

获取相关字段。

User.fetch_related("emails", "manager")

参数:¶

*args¶

应获取的相关字段。

using_db=None¶

特定数据库连接，用于替代默认绑定

返回类型:¶

无

classmethod filter(*args, **kwargs)[source]¶

生成应用了过滤器的 QuerySet。

参数:¶

*args¶

包含约束的 Q 函数。将进行 AND 操作。

**kwargs¶

简单筛选约束。

返回类型:¶

QuerySet[typing_extensions.Self]

classmethod first(using_db=None)[source]¶

生成返回第一条记录的 QuerySet。

返回类型:¶

QuerySetSingle[Optional[typing_extensions.Self]]

classmethod get(*args, using_db=None, **kwargs)[source]¶

使用提供的过滤器参数获取 Model 类型的一个单一记录。

user = await User.get(username="foo")

参数:¶

using_db=None¶

要使用的数据库连接

*args¶

包含约束的 Q 函数。将进行 AND 操作。

**kwargs¶

简单筛选约束。

引发:¶

• MultipleObjectsReturned – 如果提供的搜索返回多个对象。

• DoesNotExist – 如果找不到对象。

返回类型:¶

QuerySetSingle[typing_extensions.Self]

async classmethod get_or_create(defaults=None, using_db=None, **kwargs)[source]¶

如果存在，则获取对象（根据提供的参数进行筛选），否则使用任何未指定参数作为默认值创建一个实例。

参数:¶

defaults=None¶

如果无法获取，则要添加到已创建实例的默认值。

using_db=None¶

特定数据库连接，用于替代默认绑定

**kwargs¶

查询参数。

引发:¶

• IntegrityError – 如果创建失败

• TransactionManagementError – 如果事务错误

• ParamsError – 如果默认值与 kwargs 冲突

返回类型:¶

Tuple[typing_extensions.Self, bool]

classmethod get_or_none(*args, using_db=None, **kwargs)[source]¶

使用提供的筛选参数或 None 获取 Model 类型的单个记录。

user = await User.get_or_none(username="foo")

参数:¶

using_db=None¶

要使用的特定数据库连接。

*args¶

包含约束的 Q 函数。将进行 AND 操作。

**kwargs¶

简单筛选约束。

返回类型:¶

QuerySetSingle[Optional[typing_extensions.Self]]

async classmethod in_bulk(id_list, field_name='pk', using_db=None)[source]¶

返回一个字典，将给定的每个 ID 映射到具有该 ID 的对象。如果未提供 id_list，则评估整个 QuerySet。

参数:¶

id_list¶

字段值列表

field_name='pk'¶

必须是唯一字段

using_db=None¶

特定数据库连接，用于替代默认绑定

返回类型:¶

Dict[str, Model]

property pk : Any¶

模型主键的别名。在进行过滤时可以用作字段名称，例如 .filter(pk=...) 等…

返回类型:¶

Any

classmethod raw(sql, using_db=None)[source]¶

执行原始 SQL 并返回结果

result = await User.raw("select * from users where name like '%test%'")

参数:¶

using_db=None¶

要使用的特定数据库连接

sql¶

原始 sql。

返回类型:¶

RawSQLQuery

async refresh_from_db(fields=None, using_db=None)[source]¶

从数据库刷新最新数据。当此方法在没有参数的情况下被调用时，模型的所有数据库字段都会更新为当前数据库中存在的值。

user.refresh_from_db(fields=['name'])

参数:¶

fields=None¶

要刷新的特殊字段。

using_db=None¶

特定的数据库连接，用于替代默认绑定。

引发:¶

OperationalError – 如果对象从未持久化。

返回类型:¶

无

classmethod register_listener(signal, listener)[source]¶

为当前模型类注册侦听器，用于特殊信号。

参数:¶

signal¶

tortoise.signals.Signals 之一

listener¶

可调用的侦听器

引发:¶

ConfigurationError – 当侦听器不可调用时

async save(using_db=None, update_fields=None, force_create=False, force_update=False)[source]¶

创建/更新当前模型对象。

参数:¶

update_fields=None¶

如果提供，它应该是按名称排列的字段的元组/列表。

这是应该更新的字段的子集。如果需要创建对象，则将忽略 update_fields。

using_db=None¶

特定数据库连接，用于替代默认绑定

force_create=False¶

强制创建记录

force_update=False¶

强制更新记录

引发:¶

• IncompleteInstanceError – 如果模型是部分的，并且这些字段不可用于持久性。

• IntegrityError – 如果无法创建或更新模型（特别是如果已设置 force_create 或 force_update）

返回类型:¶

无

classmethod select_for_update(nowait=False, skip_locked=False, of=(), using_db=None)[source]¶

使 QuerySet 选择更新。

返回一个查询集，它将锁定行直到事务结束，在支持的数据库上生成一个 SELECT … FOR UPDATE SQL 语句。

返回类型:¶

QuerySet[typing_extensions.Self]

update_from_dict(data)[source]¶

使用提供的字典更新当前模型。这允许从字典中批量更新模型，同时确保数据类型转换发生。

这将忽略任何额外的字段，并且不会用它们更新模型，但会对错误的类型或更新多实例关系引发错误。

参数:¶

data¶

您想要以字典格式更新的参数

返回类型:¶

Model

返回:¶

当前模型实例

引发:¶

• ConfigurationError – 当尝试更新远程实例（例如反向 ForeignKey 或 ManyToMany 关系）时

• ValueError – 当传递的参数与类型不兼容时

async classmethod update_or_create(defaults=None, using_db=None, **kwargs)[source]¶

使用给定的 kwargs 更新对象，必要时创建一个新对象，这是一种便捷方法。

参数:¶

defaults=None¶

用于更新对象的默认值。

using_db=None¶

特定数据库连接，用于替代默认绑定

**kwargs¶

查询参数。

返回类型:¶

Tuple[Model, bool]

class tortoise.Tortoise[source]¶

apps : dict[str, dict[str, type[Model]]] = {}¶

async classmethod close_connections()[source]¶

干净关闭所有连接。

退出时需要调用此方法，否则事件循环可能永远无法完成，因为它正在等待连接断开。

警告

此方法已弃用，将在未来版本中删除。请改用 connections.close_all。

返回类型:¶

无

classmethod describe_model(model, serializable=True)[源代码]¶

描述给定的模型列表或所有已注册的模型。

参数:¶

model¶

要描述的模型

serializable=True¶

False 如果想要原始 Python 对象，True 如果想要可 JSON 序列化的数据。（默认为 True）

参见 tortoise.models.Model.describe()

警告

此方法已弃用，请改用 tortoise.models.Model.describe()

返回类型:¶

dict

classmethod describe_models(models=None, serializable=True)[源代码]¶

描述给定的模型列表或所有已注册的模型。

参数:¶

models=None¶

要描述的模型列表，如果未提供，则描述所有已注册的模型

serializable=True¶

False 如果想要原始 Python 对象，True 如果想要可 JSON 序列化的数据。（默认为 True）

返回类型:¶

Dict[str, dict]

返回:¶

包含模型限定符作为键的字典，以及与 describe_model(...) 相同的输出作为值

{
    "models.User": {...},
    "models.Permission": {...}
}

async classmethod generate_schemas(safe=True)[source]¶

根据提供给 .init() 方法的模型生成模式。如果模式已存在，则会失败，因此不建议将其用作应用程序工作流的一部分

参数:¶

safe=True¶

设置成 true 时，仅在表尚不存在时创建表。

引发:¶

ConfigurationError – 当 .init() 未被调用时。

返回类型:¶

无

classmethod get_connection(connection_name)[source]¶

按名称返回连接。

引发:¶

ConfigurationError – 如果连接名称不存在。

警告

此项已弃用，将在未来版本中移除。请使用 connections.get 代替。

返回类型:¶

BaseDBAsyncClient

async classmethod init(config=None, config_file=None, _create_db=False, db_url=None, modules=None, use_tz=False, timezone='UTC', routers=None)[source]¶

设置 Tortoise-ORM。

您只能使用 config、config_file 和 (db_url, modules) 中的一种进行配置。

参数:¶

config=None¶

包含配置的字典

示例

{
    'connections': {
        # Dict format for connection
        'default': {
            'engine': 'tortoise.backends.asyncpg',
            'credentials': {
                'host': 'localhost',
                'port': '5432',
                'user': 'tortoise',
                'password': 'qwerty123',
                'database': 'test',
            }
        },
        # Using a DB_URL string
        'default': 'postgres://postgres:qwerty123@localhost:5432/test'
    },
    'apps': {
        'my_app': {
            'models': ['__main__'],
            # If no default_connection specified, defaults to 'default'
            'default_connection': 'default',
        }
    },
    'routers': ['path.router1', 'path.router2'],
    'use_tz': False,
    'timezone': 'UTC'
}

config_file=None¶

指向包含与上述格式相同的配置的 .json 或 .yml（如果已安装 PyYAML）文件的路径。

db_url=None¶

使用 DB_URL 字符串。请参阅 DB_URL

modules=None¶

定义“应用”和应为模型发现的模块的 key: [list_of_modules] 字典。

_create_db=False¶

如果 True 尝试为指定的连接创建数据库，可用于测试目的。

use_tz=False¶

指定 datetime 默认是否支持时区的布尔值。

timezone='UTC'¶

要使用的时区，默认为 UTC。

routers=None¶

db 路由器 str 路径或模块的列表。

引发:¶

ConfigurationError – 对于任何配置错误

返回类型:¶

无

类方法 init_models(models_paths, app_label, _init_relations=True)[源代码]¶

Tortoise ORM 模型的早期初始化。

初始化模型之间的关系。这不会初始化任何数据库连接。

参数:¶

models_paths¶

要初始化的模型路径

app_label¶

应用程序标签，例如“models”

_init_relations=True¶

是否初始化关系

引发:¶

ConfigurationError – 如果模型无效。

返回类型:¶

无