Arangodantic

Database models for ArangoDB using Pydantic base models.

Installation

The package is available on PyPI:

pip install arangodantic

# Or with Shylock
pip install arangodantic[shylock]

Usage

Define your database models by extending either DocumentModel or EdgeModel. Nested structures can be created by extending pydantic.BaseModel.

Configure Arangodantic. You can optionally define a collection name prefix, a key generation function and a lock (needed if you want to use the locking functionality; Shylock is supported out of the box, any other locking service such as Sherlock should at least in theory also work).

Ensure you have an ArangoDB server available with known credentials

docker run --rm -p 8529:8529 -e ARANGO_ROOT_PASSWORD="" arangodb/arangodb:3.7.2.1

import asyncio
from uuid import uuid4

from aioarangodb import ArangoClient
from pydantic import BaseModel
from shylock import AsyncLock as Lock
from shylock import ShylockAioArangoDBBackend
from shylock import configure as configure_shylock

from arangodantic import ASCENDING, DocumentModel, EdgeModel, configure


# Define models
class Owner(BaseModel):
    """Dummy owner Pydantic model."""

    first_name: str
    last_name: str


class Company(DocumentModel):
    """Dummy company Arangodantic model."""

    company_id: str
    owner: Owner


class Link(EdgeModel):
    """Dummy Link Arangodantic model."""

    type: str


async def main():
    # Configure the database settings
    hosts = "http://localhost:8529"
    username = "root"
    password = ""
    database = "example"
    prefix = "example-"

    client = ArangoClient(hosts=hosts)
    # Connect to "_system" database and create the actual database if it doesn't exist
    # Only for demo, you likely want to create the database in advance.
    sys_db = await client.db("_system", username=username, password=password)
    if not await sys_db.has_database(database):
        await sys_db.create_database(database)

    # Configure Arangodantic and Shylock
    db = await client.db(database, username=username, password=password)
    configure_shylock(await ShylockAioArangoDBBackend.create(db, f"{prefix}shylock"))
    configure(db, prefix=prefix, key_gen=uuid4, lock=Lock)

    # Create collections if they don't yet exist
    # Only for demo, you likely want to create the collections in advance.
    await Company.ensure_collection()
    await Link.ensure_collection()

    # Let's create some example entries
    owner = Owner(first_name="John", last_name="Doe")
    company = Company(company_id="1234567-8", owner=owner)
    await company.save()
    print(f"Company saved with key: {company.key_}")

    second_owner = Owner(first_name="Jane", last_name="Doe")
    second_company = Company(company_id="2345678-9", owner=second_owner)
    await second_company.save()
    print(f"Second company saved with key: {second_company.key_}")

    link = Link(_from=company, _to=second_company, type="CustomerOf")
    await link.save()
    print(f"Link saved with key: {link.key_}")

    # Hold named locks while loading and doing changes
    async with Company.lock_and_load(company.key_) as c:
        assert c.owner == owner
        c.owner.first_name = "Joanne"
        await c.save()

    await company.reload()
    print(f"Updated owner of company to '{company.owner!r}'")

    # Let's explore the find functionality
    # Note: You likely want to add indexes to support the queries
    print("Finding companies owned by a person with last name 'Doe'")
    async with (await Company.find({"owner.last_name": "Doe"}, count=True)) as cursor:
        print(f"Found {len(cursor)} companies")
        async for found_company in cursor:
            print(f"Company: {found_company.company_id}")

    # Supported operators include: "==", "!=", "<", "<=", ">", ">="
    found_company = await Company.find_one(
        {"owner.last_name": "Doe", "_id": {"!=": company}}
    )
    print(f"Found the company {found_company.key_}")

    # Find also supports sorting and the cursor can easily be converted to a list
    companies = await (
        await Company.find(
            sort=[
                ("owner.last_name", ASCENDING),
                ("owner.first_name", ASCENDING),
            ]
        )
    ).to_list()
    print("Companies sorted by owner:")
    for c in companies:
        print(f"Company {c.company_id}, owner: {c.owner!r}")


if __name__ == "__main__":
    # Starting from Python 3.7 ->
    # asyncio.run(main())

    # Compatible with Python 3.6 ->
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())

You might find migrate-anything useful for creating and managing collections and indexes.

More examples

• The graph example shows how arangodantic can be used with graphs. Please note that graph related functionality is at the moment really limited and will likely be extended later and might even be restructured, so use with caution.

License

This code is released under the BSD 3-Clause license. Details in the LICENSE file.