ArangoDB Arango ORM到Pydantic

4zcjmb1e  于 2024-01-04  发布在  Go
关注(0)|答案(1)|浏览(186)

我正在把我的代码从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,您可以轻松定义顶点和边模型:

  1. from typing import Annotated
  2. import datetime
  3. from pydango.indexes import PersistentIndex
  4. from pydango import (
  5. VertexModel,
  6. EdgeModel,
  7. EdgeCollectionConfig,
  8. VertexCollectionConfig,
  9. Relation,
  10. )
  11. class Visited(EdgeModel):
  12. rating: int
  13. on_date: datetime.date
  14. class Collection(EdgeCollectionConfig):
  15. name = "visited"
  16. indexes = [
  17. PersistentIndex(fields=["rating"]),
  18. ]
  19. class LivesIn(EdgeModel):
  20. since: datetime.datetime
  21. class Collection(EdgeCollectionConfig):
  22. name = "lives_in"
  23. class City(VertexModel):
  24. name: str
  25. population: int
  26. class Collection(VertexCollectionConfig):
  27. name = "cities"
  28. indexes = [PersistentIndex(fields=["name"])]
  29. class Person(VertexModel):
  30. name: str
  31. age: int
  32. lives_in: Annotated[City, Relation[LivesIn]]
  33. visited: Annotated[list[City], Relation[Visited]]
  34. class Collection(VertexCollectionConfig):
  35. name = "people"
  36. indexes = [
  37. PersistentIndex(fields=["name"]),
  38. PersistentIndex(fields=["age"]),
  39. ]

字符串

CRUD操作

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

  1. # Create a new person
  2. async def async_application():
  3. person = Person(name="Alice", age=30)
  4. person.lives_in = City(name="Buenos Aires", population=16_500_000)
  5. person.visited = [City(name="San Francisco", population=800_000)]
  6. person.edges.lives_in = LivesIn(since=datetime.datetime.now())
  7. person.edges.visited = [Visited(rating=5, on_date=datetime.date.today())]
  8. await session.save(person)
  9. # Read a person by their ID
  10. retrieved_person = await session.get(Person, person.id)
  11. # Update the person's age
  12. person.age = 31
  13. await session.save(person)

正在运行缓存

简单查询

构造并执行一个简单的查询来检索所有年龄在25岁以上的人:

  1. from pydango.orm import for_
  2. query = for_(Person).filter(Person.age > 25).return_(Person)
  3. people_over_25 = await session.execute(query)

业务查询

构造并执行一个简单的查询,查询访问同一个城市的人访问过的城市:

  1. from pydango.orm import traverse
  2. from pydango import TraversalDirection
  3. person_visited_cities = traverse(
  4. Person,
  5. edges=[Person.visited],
  6. start=person.id,
  7. depth=(1, 2),
  8. direction=TraversalDirection.INBOUND,
  9. ).return_(Person)


如果你愿意为你的项目尝试给予PydangORM,我会很高兴。你的反馈将是无价的,如果你有兴趣贡献或有任何问题,我很乐意合作。

展开查看全部

相关问题