Ormar

概述

ormar 包是 Python 的异步 ORM，支持 Postgres、MySQL 和 SQLite。

使用 ormar 的主要好处是：

• 获取可与异步框架（fastapi、starlette 等）一起使用的异步 ORM
• 只需维护一个模型 - 您不必维护 pydantic 和其他 orm 模型（sqlalchemy、peewee、gino 等）

目标是创建一个简单的 ORM，可以直接与基于 pydantic 的数据验证的 fastapi 一起使用（作为请求和响应模型）。

Ormar——除了名字中明显的“ORM”之外——它的名字来源于瑞典语中的ormar，意思是蛇，以及克罗地亚语中的ormar(e)，意思是内阁。

对于 python ORM 来说，还有什么比 Snakes Cabinet 更好的名字呢:)

如果你喜欢 ormar，记得给 github 中的存储库加注星标！

我们建立的社区越大，就越容易发现错误并吸引贡献者；）

文档

查看文档了解详细信息。

请注意，为了简洁起见，大多数文档片段省略了数据库的创建和调度异步运行函数的执行。

如果您想要比文档中更多的现实生活示例，您可以查看测试文件夹，因为它们实际上必须在大多数测试中创建并连接到数据库。

但请记住，这些只是测试，并非所有解决方案都适合在现实生活中使用。

fastapi 生态系统的一部分

作为 fastapi 生态系统的一部分，ormar 在以某种方式与数据库一起工作的选定库中受到支持。

Ormar 仍然与 sql 方言无关 - 因此仅实现在所有受支持的后端中工作的列。

作为 ormar 的扩展，实现特定方言的列相对容易。

Postgres 特定列实现：ormar-postgres-extensions

如果您维护或使用不同的库并希望它支持 ormar，请告诉我们如何提供帮助。

依赖关系

Ormar 的构建有：

• 用于构建查询的 sqlalchemy 核心。
• 用于跨数据库异步支持的数据库。
• pydantic 用于数据验证。

执照

ormar 是作为开源软件构建的，并将保持完全免费（MIT 许可证）。

当我编写开源代码来解决工作中的日常问题或促进和建立强大的 Python 社区时，您可以说谢谢并请我喝杯咖啡或每月资助我，以帮助确保我的工作保持免费和维护。

赞助

从 sqlalchemy 和现有数据库迁移

如果您当前使用 sqlalchemy 并想切换到 ormar，请查看自动翻译工具，它可以帮助您翻译现有的 sqlalchemy orm 模型，这样您就不必手动执行此操作。

Beta 版本可在 github 上找到：sqlalchemy-to-ormar 或简单地 pip install sqlalchemy-to-ormar

sqlalchemy-to-ormar 可以与 sqlacodegen 配合使用，从现有数据库自动映射/生成 ormar 模型，即使您的项目不使用 sqlalchemy。

迁移和数据库创建

因为 ormar 是基于 SQLAlchemy 核心构建的，所以您可以使用 alembic 来提供数据库迁移（对于生产代码，您确实应该这样做）。

对于测试和基本应用程序，sqlalchemy 已经足够了：

# note this is just a partial snippet full working example below
# 1. Imports
import sqlalchemy
import databases
import ormar

# 2. Initialization
DATABASE_URL = "sqlite:///db.sqlite"
base_ormar_config = ormar.OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database=databases.Database(DATABASE_URL),
    engine=sqlalchemy.create_engine(DATABASE_URL),
)

# Define models here

# 3. Database creation and tables creation
base_ormar_config.metadata.create_all(engine)

有关 alembic 的示例配置以及有关迁移和数据库创建的更多信息，请访问迁移文档部分。

封装版本

ormar 仍在开发中：我们建议固定任何依赖项（即 ormar~=0.9.1）

ormar 还遵循发布编号，即重大更改会增加主要版本号，而其他更改和修复会增加次要版本号，因此对于后者，您应该可以安全地更新，但请务必先阅读发行版文档。 example: (0.5.2 -> 0.6.0 - breaking, 0.5.2 -> 0.5.3 - non breaking) 。

异步Python

请注意，ormar 是一个异步 ORM，这意味着您必须等待对计划在事件循环中执行的方法的调用。 Python 有一个内置模块 asyncio，可以让你做到这一点。

请注意，大多数“普通”Python 解释器不允许在函数外部执行 wait（因为您实际上安排此函数延迟执行，并且不会立即获得结果）。

在现代 Web 框架（如 fastapi）中，框架将为您处理此问题，但如果您打算自己执行此操作，则需要手动执行此操作，如下面的快速入门中所述。

快速入门

请注意，您可以在 github 上的示例文件夹中找到相同的脚本。

from typing import Optional

import databases

import ormar
import sqlalchemy

DATABASE_URL = "sqlite:///db.sqlite"
base_ormar_config = ormar.OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database=databases.Database(DATABASE_URL),
    engine = sqlalchemy.create_engine(DATABASE_URL),
)

# note that this step is optional -> all ormar cares is a field with name
# ormar_config # and proper parameters, but this way you do not have to repeat
# the same parameters if you use only one database
#
# Note that all type hints are optional
# below is a perfectly valid model declaration
# class Author(ormar.Model):
#     ormar_config = base_ormar_config.copy(tablename="authors")
#
#     id = ormar.Integer(primary_key=True) # <= notice no field types
#     name = ormar.String(max_length=100)


class Author(ormar.Model):
    ormar_config = base_ormar_config.copy(tablename="authors")

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)


class Book(ormar.Model):
    ormar_config = base_ormar_config.copy(tablename="books")

    id: int = ormar.Integer(primary_key=True)
    author: Optional[Author] = ormar.ForeignKey(Author)
    title: str = ormar.String(max_length=100)
    year: int = ormar.Integer(nullable=True)


# create the database
# note that in production you should use migrations
# note that this is not required if you connect to existing database
# just to be sure we clear the db before
base_ormar_config.metadata.drop_all(engine)
base_ormar_config.metadata.create_all(engine)


# all functions below are divided into functionality categories
# note how all functions are defined with async - hence can use await AND needs to
# be awaited on their own
async def create():
    # Create some records to work with through QuerySet.create method.
    # Note that queryset is exposed on each Model's class as objects
    tolkien = await Author.objects.create(name="J.R.R. Tolkien")
    await Book.objects.create(author=tolkien, title="The Hobbit", year=1937)
    await Book.objects.create(author=tolkien, title="The Lord of the Rings", year=1955)
    await Book.objects.create(author=tolkien, title="The Silmarillion", year=1977)

    # alternative creation of object divided into 2 steps
    sapkowski = Author(name="Andrzej Sapkowski")
    # do some stuff
    await sapkowski.save()

    # or save() after initialization
    await Book(author=sapkowski, title="The Witcher", year=1990).save()
    await Book(author=sapkowski, title="The Tower of Fools", year=2002).save()

    # to read more about inserting data into the database
    # visit: https://collerek.github.io/ormar/queries/create/


async def read():
    # Fetch an instance, without loading a foreign key relationship on it.
    # Django style
    book = await Book.objects.get(title="The Hobbit")
    # or python style
    book = await Book.objects.get(Book.title == "The Hobbit")
    book2 = await Book.objects.first()

    # first() fetch the instance with lower primary key value
    assert book == book2

    # you can access all fields on loaded model
    assert book.title == "The Hobbit"
    assert book.year == 1937

    # when no condition is passed to get()
    # it behaves as last() based on primary key column
    book3 = await Book.objects.get()
    assert book3.title == "The Tower of Fools"

    # When you have a relation, ormar always defines a related model for you
    # even when all you loaded is a foreign key value like in this example
    assert isinstance(book.author, Author)
    # primary key is populated from foreign key stored in books table
    assert book.author.pk == 1
    # since the related model was not loaded all other fields are None
    assert book.author.name is None

    # Load the relationship from the database when you already have the related model
    # alternatively see joins section below
    await book.author.load()
    assert book.author.name == "J.R.R. Tolkien"

    # get all rows for given model
    authors = await Author.objects.all()
    assert len(authors) == 2

    # to read more about reading data from the database
    # visit: https://collerek.github.io/ormar/queries/read/


async def update():
    # read existing row from db
    tolkien = await Author.objects.get(name="J.R.R. Tolkien")
    assert tolkien.name == "J.R.R. Tolkien"
    tolkien_id = tolkien.id

    # change the selected property
    tolkien.name = "John Ronald Reuel Tolkien"
    # call update on a model instance
    await tolkien.update()

    # confirm that object was updated
    tolkien = await Author.objects.get(name="John Ronald Reuel Tolkien")
    assert tolkien.name == "John Ronald Reuel Tolkien"
    assert tolkien.id == tolkien_id

    # alternatively update data without loading
    await Author.objects.filter(name__contains="Tolkien").update(name="J.R.R. Tolkien")

    # to read more about updating data in the database
    # visit: https://collerek.github.io/ormar/queries/update/


async def delete():
    silmarillion = await Book.objects.get(year=1977)
    # call delete() on instance
    await silmarillion.delete()

    # alternatively delete without loading
    await Book.objects.delete(title="The Tower of Fools")

    # note that when there is no record ormar raises NoMatch exception
    try:
        await Book.objects.get(year=1977)
    except ormar.NoMatch:
        print("No book from 1977!")

    # to read more about deleting data from the database
    # visit: https://collerek.github.io/ormar/queries/delete/

    # note that despite the fact that record no longer exists in database
    # the object above is still accessible and you can use it (and i.e. save()) again.
    tolkien = silmarillion.author
    await Book.objects.create(author=tolkien, title="The Silmarillion", year=1977)


async def joins():
    # Tho join two models use select_related

    # Django style
    book = await Book.objects.select_related("author").get(title="The Hobbit")
    # Python style
    book = await Book.objects.select_related(Book.author).get(
        Book.title == "The Hobbit"
    )

    # now the author is already prefetched
    assert book.author.name == "J.R.R. Tolkien"

    # By default you also get a second side of the relation
    # constructed as lowercase source model name +'s' (books in this case)
    # you can also provide custom name with parameter related_name

    # Django style
    author = await Author.objects.select_related("books").all(name="J.R.R. Tolkien")
    # Python style
    author = await Author.objects.select_related(Author.books).all(
        Author.name == "J.R.R. Tolkien"
    )
    assert len(author[0].books) == 3

    # for reverse and many to many relations you can also prefetch_related
    # that executes a separate query for each of related models

    # Django style
    author = await Author.objects.prefetch_related("books").get(name="J.R.R. Tolkien")
    # Python style
    author = await Author.objects.prefetch_related(Author.books).get(
        Author.name == "J.R.R. Tolkien"
    )
    assert len(author.books) == 3

    # to read more about relations
    # visit: https://collerek.github.io/ormar/relations/

    # to read more about joins and subqueries
    # visit: https://collerek.github.io/ormar/queries/joins-and-subqueries/


async def filter_and_sort():
    # to filter the query you can use filter() or pass key-value pars to
    # get(), all() etc.
    # to use special methods or access related model fields use double
    # underscore like to filter by the name of the author use author__name
    # Django style
    books = await Book.objects.all(author__name="J.R.R. Tolkien")
    # python style
    books = await Book.objects.all(Book.author.name == "J.R.R. Tolkien")
    assert len(books) == 3

    # filter can accept special methods also separated with double underscore
    # to issue sql query ` where authors.name like "%tolkien%"` that is not
    # case sensitive (hence small t in Tolkien)
    # Django style
    books = await Book.objects.filter(author__name__icontains="tolkien").all()
    # python style
    books = await Book.objects.filter(Book.author.name.icontains("tolkien")).all()
    assert len(books) == 3

    # to sort use order_by() function of queryset
    # to sort decreasing use hyphen before the field name
    # same as with filter you can use double underscores to access related fields
    # Django style
    books = (
        await Book.objects.filter(author__name__icontains="tolkien")
        .order_by("-year")
        .all()
    )
    # python style
    books = (
        await Book.objects.filter(Book.author.name.icontains("tolkien"))
        .order_by(Book.year.desc())
        .all()
    )
    assert len(books) == 3
    assert books[0].title == "The Silmarillion"
    assert books[2].title == "The Hobbit"

    # to read more about filtering and ordering
    # visit: https://collerek.github.io/ormar/queries/filter-and-sort/


async def subset_of_columns():
    # to exclude some columns from loading when querying the database
    # you can use fileds() method
    hobbit = await Book.objects.fields(["title"]).get(title="The Hobbit")
    # note that fields not included in fields are empty (set to None)
    assert hobbit.year is None
    assert hobbit.author is None

    # selected field is there
    assert hobbit.title == "The Hobbit"

    # alternatively you can provide columns you want to exclude
    hobbit = await Book.objects.exclude_fields(["year"]).get(title="The Hobbit")
    # year is still not set
    assert hobbit.year is None
    # but author is back
    assert hobbit.author is not None

    # also you cannot exclude primary key column - it's always there
    # even if you EXPLICITLY exclude it it will be there

    # note that each model have a shortcut for primary_key column which is pk
    # and you can filter/access/set the values by this alias like below
    assert hobbit.pk is not None

    # note that you cannot exclude fields that are not nullable
    # (required) in model definition
    try:
        await Book.objects.exclude_fields(["title"]).get(title="The Hobbit")
    except pydantic.ValidationError:
        print("Cannot exclude non nullable field title")

    # to read more about selecting subset of columns
    # visit: https://collerek.github.io/ormar/queries/select-columns/


async def pagination():
    # to limit number of returned rows use limit()
    books = await Book.objects.limit(1).all()
    assert len(books) == 1
    assert books[0].title == "The Hobbit"

    # to offset number of returned rows use offset()
    books = await Book.objects.limit(1).offset(1).all()
    assert len(books) == 1
    assert books[0].title == "The Lord of the Rings"

    # alternatively use paginate that combines both
    books = await Book.objects.paginate(page=2, page_size=2).all()
    assert len(books) == 2
    # note that we removed one book of Sapkowski in delete()
    # and recreated The Silmarillion - by default when no order_by is set
    # ordering sorts by primary_key column
    assert books[0].title == "The Witcher"
    assert books[1].title == "The Silmarillion"

    # to read more about pagination and number of rows
    # visit: https://collerek.github.io/ormar/queries/pagination-and-rows-number/


async def aggregations():
    # count:
    assert 2 == await Author.objects.count()

    # exists
    assert await Book.objects.filter(title="The Hobbit").exists()

    # maximum
    assert 1990 == await Book.objects.max(columns=["year"])

    # minimum
    assert 1937 == await Book.objects.min(columns=["year"])

    # average
    assert 1964.75 == await Book.objects.avg(columns=["year"])

    # sum
    assert 7859 == await Book.objects.sum(columns=["year"])

    # to read more about aggregated functions
    # visit: https://collerek.github.io/ormar/queries/aggregations/


async def raw_data():
    # extract raw data in a form of dicts or tuples
    # note that this skips the validation(!) as models are
    # not created from parsed data

    # get list of objects as dicts
    assert await Book.objects.values() == [
        {"id": 1, "author": 1, "title": "The Hobbit", "year": 1937},
        {"id": 2, "author": 1, "title": "The Lord of the Rings", "year": 1955},
        {"id": 4, "author": 2, "title": "The Witcher", "year": 1990},
        {"id": 5, "author": 1, "title": "The Silmarillion", "year": 1977},
    ]

    # get list of objects as tuples
    assert await Book.objects.values_list() == [
        (1, 1, "The Hobbit", 1937),
        (2, 1, "The Lord of the Rings", 1955),
        (4, 2, "The Witcher", 1990),
        (5, 1, "The Silmarillion", 1977),
    ]

    # filter data - note how you always get a list
    assert await Book.objects.filter(title="The Hobbit").values() == [
        {"id": 1, "author": 1, "title": "The Hobbit", "year": 1937}
    ]

    # select only wanted fields
    assert await Book.objects.filter(title="The Hobbit").values(["id", "title"]) == [
        {"id": 1, "title": "The Hobbit"}
    ]

    # if you select only one column you could flatten it with values_list
    assert await Book.objects.values_list("title", flatten=True) == [
        "The Hobbit",
        "The Lord of the Rings",
        "The Witcher",
        "The Silmarillion",
    ]

    # to read more about extracting raw values
    # visit: https://collerek.github.io/ormar/queries/aggregations/


async def with_connect(function):
    # note that for any other backend than sqlite you actually need to
    # connect to the database to perform db operations
    async with database:
        await function()

    # note that if you use framework like `fastapi` you shouldn't connect
    # in your endpoints but have a global connection pool
    # check https://collerek.github.io/ormar/fastapi/ and section with db connection


# gather and execute all functions
# note - normally import should be at the beginning of the file
import asyncio

# note that normally you use gather() function to run several functions
# concurrently but we actually modify the data and we rely on the order of functions
for func in [
    create,
    read,
    update,
    delete,
    joins,
    filter_and_sort,
    subset_of_columns,
    pagination,
    aggregations,
    raw_data,
]:
    print(f"Executing: {func.__name__}")
    asyncio.run(with_connect(func))

# drop the database tables
metadata.drop_all(engine)

奥马尔规格

查询集方法

• 创建（**kwargs）：-> 模型
• get(*args, **kwargs): -> Model
• get_or_none(*args, **kwargs): -> Optional[Model]
• get_or_create(_defaults: Optional[Dict[str, Any]] = None, *args, **kwargs) -> Tuple[Model, bool]
• first(*args, **kwargs): -> Model
• update(each: bool = False, **kwargs) -> int
• update_or_create(**kwargs) -> Model
• bulk_create(objects: List[Model]) -> None
• bulk_update(objects: List[Model], columns: List[str] = None) -> None
• delete(*args, each: bool = False, **kwargs) -> int
• all(*args, **kwargs) -> List[Optional[Model]]
• iterate(*args, **kwargs) -> AsyncGenerator[Model]
• filter(*args, **kwargs) -> QuerySet
• exclude(*args, **kwargs) -> QuerySet
• select_related(related: Union[List, str]) -> QuerySet
• prefetch_related(related: Union[List, str]) -> QuerySet
• limit(limit_count: int) -> QuerySet
• offset(offset: int) -> QuerySet
• count(distinct: bool = True) -> int
• 存在() -> 布尔值
• max(columns: List[str]) -> Any
• min(columns: List[str]) -> Any
• avg(columns: List[str]) -> Any
• sum(columns: List[str]) -> Any
• fields(columns: Union[List, str, set, dict]) -> QuerySet
• exclude_fields(columns: Union[List, str, set, dict]) -> QuerySet
• order_by(columns:Union[List, str]) -> QuerySet
• values(fields: Union[List, str, Set, Dict])
• values_list(fields: Union[List, str, Set, Dict])

关系类型

• 一对多 - 使用ForeignKey(to: Model)
• 多对多 - 与 ManyToMany(to: Model, Optional[through]: Model)

模型字段类型

可用模型字段（具有必需的参数 - 文档中的可选参数）：

• 字符串（最大长度）
• 文本（）
• 布尔值()
• 整数()
• 漂浮（）
• 日期（）
• 时间（）
• 日期时间()
• JSON()
• 大整数()
• 小整数()
• 小数（小数位数、精度）
• UUID()
• 大二进制（最大长度）
• 枚举（枚举类）
• 类似于 Field 的枚举 - 通过将选择传递给任何其他 Field 类型
• EncryptedString - 通过传递 encrypt_secret 和 encrypt_backend
• 外键（至）
• 多对多（到）

可用字段选项

所有字段类型都支持以下关键字参数。

• 主键：布尔值
• 可空：布尔值
• 默认值：任何
• 服务器默认值：任何
• 索引：布尔值
• 唯一：布尔
• 名称：str

除非设置了以下其中一项，否则所有字段均为必填：

• nullable - 创建一个可为空的列。将默认值设置为 False。详细信息请阅读字段通用参数。
• sql_nullable - 用于为 pydantic 和数据库设置不同的设置。将默认值设置为可为空的值。详细信息请阅读字段通用参数。
• 默认 - 设置字段的默认值。不适用于关系字段
• server_default - 设置服务器端字段的默认值（如 sqlalchemy 的 func.now()）。不适用于关系字段
• 带自动增量的主键 - 当某列设置为主键并在此列上设置自动增量时。默认情况下，int 主键设置自动增量。

可用信号

信号允许针对给定模型上的给定事件触发您的函数。

• 预保存
• 保存后
• 预更新
• 更新后
• 预删除
• 删除后
• 预关系添加
• 后关系添加
• 预关系删除
• 删除关系后