ArangoDB Arango ORM到Pydantic

4zcjmb1e  于 12个月前  发布在  Go
关注(0)|答案(1)|浏览(151)

我正在把我的代码从arango-cloud转移到pydantic上,我的问题是如何在pydantic中编写集合类,图类,边类,关系类。
请各位指导。谢谢。
我还尝试安装arangodantic,结果出现错误:Error

bsxbgnwa

bsxbgnwa1#

TMalik
我知道你正在寻找一个基于Pydantic的ArangoDB ORM作为arangodantic的替代品,你遇到了问题。我在我的项目中也有类似的需求,我需要一个基于PydanticArangoDB ORM,可以与FastAPI无缝集成。
在研究并发现现有解决方案的局限性后,例如:

我开发了PydangORM。这个框架旨在提供更直观和类似SQLAchemy的体验,用于处理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操作

使用模型执行基本的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,我会很高兴。你的反馈将是无价的,如果你有兴趣贡献或有任何问题,我很乐意合作。

相关问题