Pydastic

Pydastic is an elasticsearch python ORM based on Pydantic.

Pip:

pip install pydastic

Poetry:

poetry add pydastic

• Simple CRUD operations supported
• Sessions for simplifying bulk operations (a la SQLAlchemy)
• Dynamic index support when committing operations

• Search API (for now, exposed client can be used)

from pydastic import ESModel

class User(ESModel):
    name: str
    phone: Optional[str]
    last_login: datetime = Field(default_factory=datetime.now)

    class Meta:
        index = "user"

An elasticsearch connection can be setup by using the connect function. This function adopts the same signature as the elasticsearch.Elasticsearch client and supports editor autocomplete. Make sure to call this only once. No protection is put in place against multiple calls, might affect performance negatively.

from pydastic import connect

connect(hosts="localhost:9200")

# Create and save doc
user = User(name="John", age=20)
user.save(wait_for=True)  # wait_for explained below

assert user.id != None

# Update doc
user.name = "Sam"
user.save(wait_for=True)

got = User.get(id=user.id)
assert got == user

user = User(name="Marie")
user.save(wait_for=True)

user.delete(wait_for=True)

Sessions are inspired by SQL Alchemy's sessions, and are used for simplifying bulk operations using the Elasticsearch client. From what I've seen, the ES client makes it pretty hard to use the bulk API, so they created bulk helpers (which in turn have incomplete/wrong docs).

john = User(name="John")
sarah = User(name="Sarah")

with Session() as session:
    session.save(john)
    session.save(sarah)
    session.commit()

With an ORM, bulk operations can be exposed neatly through a simple API. Pydastic also offers more informative errors on issues encountered during bulk operations. This is possible by suppressing the built-in elastic client errors and extracting more verbose ones instead.

Example error:

pydastic.error.BulkError: [
    {
        "update": {
            "_index": "user",
            "_type": "_doc",
            "_id": "test",
            "status": 404,
            "error": {
                "type": "document_missing_exception",
                "reason": "[_doc][test]: document missing",
                "index_uuid": "cKD0254aQRWF-E2TMxHa4Q",
                "shard": "0",
                "index": "user"
            }
        }
    },
    {
        "update": {
            "_index": "user",
            "_type": "_doc",
            "_id": "test2",
            "status": 404,
            "error": {
                "type": "document_missing_exception",
                "reason": "[_doc][test2]: document missing",
                "index_uuid": "cKD0254aQRWF-E2TMxHa4Q",
                "shard": "0",
                "index": "user"
            }
        }
    }
]

The sessions API will also be available through a context manager before the v1.0 release.

Pydastic also supports dynamic index specification. The model Metaclass index definition is still mandatory, but if an index is specified when performing operations, that will be used instead. The model Metaclass index is technically a fallback, although most users will probably be using a single index per model. For some users, multiple indices per model are needed (for example in B2B businesses, one user index per client/company is needed to keep the data separated between clients).

user = User(name="Marie")
user.save(index="my-user", wait_for=True)

user.delete(index="my-user", wait_for=True)

Still haven't got an idea on how to wrap the underlying API productively. Unless I create a DSL from scratch or use elasticsearch-dsl (which I don't like due to lacking documentation), I can't really provide any value on top of the client's built-in search API. Give this a minute of thought and shoot me your suggestions if you come up with anything!

When writing tests with Pydastic (even applies when writing tests with the elasticsearch client), remember to use the wait_for=True argument when executing operations. If this is not used, then the test will continue executing even if Elasticsearch hasn't propagated the change to all nodes, giving you weird results.

For example if you save a document, then try getting it directly after, you'll get a document not found error. This is solved by using the wait_for argument in Pydastic (equivalent to refresh="wait_for" in Elasticsearch)

Here is a reference to where this argument is listed in the docs.

It's also supported in the bulk helpers even though its not mentioned in their docs, but you wouldn't figure that out unless you dug into their source and traced back several function calls where *args **kwargs are just being forwarded across calls.. :)

Part of the build flow is running the tests using elasticsearch 7.12.0 DB as well as a 7.12.0 elasticsearch-python client. Another part is using 8.1.2 as well (DB as well as client, as part of a build matrix). This ensures support for multiple versions.

None yet.

You can see the list of available releases on the GitHub Releases page.

We follow Semantic Versions specification.

We use Release Drafter. As pull requests are merged, a draft release is kept up-to-date listing the changes, ready to publish when you’re ready. With the categories option, you can categorize pull requests in release notes using labels.

This project is licensed under the terms of the MIT license. See LICENSE for more details.

@misc{pydastic,
  author = {Rami Awar},
  title = {Pydastic is an elasticsearch python ORM based on Pydantic.},
  year = {2022},
  publisher = {GitHub},
  journal = {GitHub repository},
  howpublished = {\url{https://github.com/ramiawar/pydastic}}
}