Arango ORM 到 Pydantic

Question

我正在将我的代码从 arango-orm 转移到 pydantic。我的问题是如何在 pydantic 中编写集合、图、边、关系类。

请指导。 谢谢你。

我还尝试安装

arangodantic

，结果出现错误：

Answer 1

嗨TMalik，

我了解您正在寻找基于 Pydantic 的 ArangoDB ORM 作为您遇到问题的

arangodantic

的替代品。我在我的项目中也有类似的需求，我需要一个基于 async Pydantic 的 ArangoDB ORM，它可以与 FastAPI 无缝集成。

研究并发现现有解决方案的局限性后，例如：

• arango-orm（基于marshmellow）
• arangodantic（不像 SQLAlchemy）
• 芳香（未维护/生产就绪）

我开发了PydangORM。该框架旨在提供更直观且类似 SQLAlchemy 的体验，用于处理 ArangoDB 中的集合、图形、边和关系，同时利用 Pydantic 的优势。

PydangORM 目前基于 Pydantic V1，我计划很快将其升级到 V2。以下是 PydangORM 如何使用 Pydantic 模型简化 ArangoDB 的使用：

定义模型

使用

pydangorm

，您可以轻松定义顶点和边模型：

from typing import Annotated
import datetime
from pydango.indexes import PersistentIndex

from pydango import (
    VertexModel,
    EdgeModel,
    EdgeCollectionConfig,
    VertexCollectionConfig,
    Relation,
)


class Visited(EdgeModel):
    rating: int
    on_date: datetime.date

    class Collection(EdgeCollectionConfig):
        name = "visited"
        indexes = [
            PersistentIndex(fields=["rating"]),
        ]


class LivesIn(EdgeModel):
    since: datetime.datetime

    class Collection(EdgeCollectionConfig):
        name = "lives_in"


class City(VertexModel):
    name: str
    population: int

    class Collection(VertexCollectionConfig):
        name = "cities"
        indexes = [PersistentIndex(fields=["name"])]


class Person(VertexModel):
    name: str
    age: int
    lives_in: Annotated[City, Relation[LivesIn]]
    visited: Annotated[list[City], Relation[Visited]]

    class Collection(VertexCollectionConfig):
        name = "people"
        indexes = [
            PersistentIndex(fields=["name"]),
            PersistentIndex(fields=["age"]),
        ]

增删改查操作

使用模型执行基本的 CRUD 操作：

# Create a new person
async def async_application():
    person = Person(name="Alice", age=30)
    person.lives_in = City(name="Buenos Aires", population=16_500_000)
    person.visited = [City(name="San Francisco", population=800_000)]
    person.edges.lives_in = LivesIn(since=datetime.datetime.now())
    person.edges.visited = [Visited(rating=5, on_date=datetime.date.today())]

    await session.save(person)

    # Read a person by their ID
    retrieved_person = await session.get(Person, person.id)

    # Update the person's age
    person.age = 31
    await session.save(person)

运行查询

简单查询

构建并执行一个简单的查询来检索所有 25 岁以上的人：

from pydango.orm import for_

query = for_(Person).filter(Person.age > 25).return_(Person)
people_over_25 = await session.execute(query)

遍历查询

构建并执行一个简单的查询，查询访问过同一城市的人所访问的城市：

from pydango.orm import traverse
from pydango import TraversalDirection

person_visited_cities = traverse(
    Person,
    edges=[Person.visited],
    start=person.id,
    depth=(1, 2),
    direction=TraversalDirection.INBOUND,
).return_(Person)

如果您在您的项目中尝试使用 PydangORM，我会很高兴。您的反馈非常宝贵，如果您有兴趣贡献或有任何疑问，我很乐意合作。

• 文档页面：https://nadobando.github.io/pydangorm
• Github 页面：https://github.com/nadobando/pydangorm