simplified examples

Author: miguelgrinberg

examples/alias_migration.py

@@ -40,7 +40,7 @@
 from fnmatch import fnmatch
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-from elasticsearch_dsl import Document, Keyword, M, connections, mapped_field
+from elasticsearch_dsl import Document, Keyword, connections, mapped_field
 
 ALIAS = "test-blog"
 PATTERN = ALIAS + "-*"
@@ -50,10 +50,10 @@ class BlogPost(Document):
     if TYPE_CHECKING:
         _id: int
 
-    title: M[str]
-    tags: M[List[str]] = mapped_field(Keyword(multi=True))
-    content: M[str]
-    published: M[Optional[datetime]]
+    title: str
+    tags: List[str] = mapped_field(Keyword())
+    content: str
+    published: Optional[datetime] = mapped_field(default=None)
 
     def is_published(self) -> bool:
         return bool(self.published and datetime.now() > self.published)
@@ -142,7 +142,6 @@ def main() -> None:
         title="Hello World!",
         tags=["testing", "dummy"],
         content=open(__file__).read(),
-        published=None,
     )
     bp.save(refresh=True)
 

examples/async/alias_migration.py

@@ -41,7 +41,7 @@
 from fnmatch import fnmatch
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-from elasticsearch_dsl import AsyncDocument, Keyword, M, async_connections, mapped_field
+from elasticsearch_dsl import AsyncDocument, Keyword, async_connections, mapped_field
 
 ALIAS = "test-blog"
 PATTERN = ALIAS + "-*"
@@ -51,10 +51,10 @@ class BlogPost(AsyncDocument):
     if TYPE_CHECKING:
         _id: int
 
-    title: M[str]
-    tags: M[List[str]] = mapped_field(Keyword(multi=True))
-    content: M[str]
-    published: M[Optional[datetime]]
+    title: str
+    tags: List[str] = mapped_field(Keyword())
+    content: str
+    published: Optional[datetime] = mapped_field(default=None)
 
     def is_published(self) -> bool:
         return bool(self.published and datetime.now() > self.published)
@@ -143,7 +143,6 @@ async def main() -> None:
         title="Hello World!",
         tags=["testing", "dummy"],
         content=open(__file__).read(),
-        published=None,
     )
     await bp.save(refresh=True)
 

examples/async/completion.py

@@ -36,7 +36,6 @@
     Completion,
     Keyword,
     Long,
-    M,
     Text,
     analyzer,
     async_connections,
@@ -54,15 +53,15 @@
 
 
 class Person(AsyncDocument):
-    name: M[str] = mapped_field(Text(fields={"keyword": Keyword()}))
-    popularity: M[int] = mapped_field(Long())
+    if TYPE_CHECKING:
+        _id: Optional[int] = mapped_field(default=None)
+
+    name: str = mapped_field(Text(fields={"keyword": Keyword()}), default="")
+    popularity: int = mapped_field(Long(), default=0)
 
     # completion field with a custom analyzer
     suggest: Dict[str, Any] = mapped_field(Completion(analyzer=ascii_fold), init=False)
 
-    if TYPE_CHECKING:
-        _id: Optional[int] = mapped_field(default=None)
-
     def clean(self) -> None:
         """
         Automatically construct the suggestion input and weight by taking all

examples/async/parent_child.py

@@ -53,7 +53,6 @@
     Join,
     Keyword,
     Long,
-    M,
     Text,
     async_connections,
     mapped_field,
@@ -65,33 +64,41 @@ class User(InnerDoc):
     Class used to represent a denormalized user stored on other objects.
     """
 
-    id: M[int] = mapped_field(Long(required=True))
-    signed_up: M[Optional[datetime]] = mapped_field(Date())
-    username: M[str] = mapped_field(Text(fields={"keyword": Keyword()}, required=True))
-    email: M[Optional[str]] = mapped_field(Text(fields={"keyword": Keyword()}))
-    location: M[Optional[str]] = mapped_field(Text(fields={"keyword": Keyword()}))
+    id: int = mapped_field(Long())
+    signed_up: Optional[datetime] = mapped_field(Date())
+    username: str = mapped_field(Text(fields={"keyword": Keyword()}))
+    email: Optional[str] = mapped_field(Text(fields={"keyword": Keyword()}))
+    location: Optional[str] = mapped_field(Text(fields={"keyword": Keyword()}))
 
 
 class Comment(InnerDoc):
     """
     Class wrapper for nested comment objects.
     """
 
-    author: M[User]
-    created: M[datetime]
-    content: M[str]
+    author: User
+    created: datetime
+    content: str
 
 
 class Post(AsyncDocument):
     """
     Base class for Question and Answer containing the common fields.
     """
 
-    author: M[User]
-    created: M[Optional[datetime]]
-    body: str
-    question_answer: M[Any] = mapped_field(Join(relations={"question": "answer"}))
-    comments: M[List[Comment]]
+    author: User
+
+    if TYPE_CHECKING:
+        _routing: str = mapped_field(default=None)
+        _index: AsyncIndex = mapped_field(default=None)
+        _id: Optional[int] = mapped_field(default=None)
+
+    created: Optional[datetime] = mapped_field(default=None)
+    body: str = mapped_field(default="")
+    comments: List[Comment] = mapped_field(default_factory=list)
+    question_answer: Any = mapped_field(
+        Join(relations={"question": "answer"}), default_factory=dict
+    )
 
     @classmethod
     def _matches(cls, hit: Dict[str, Any]) -> bool:
@@ -127,11 +134,10 @@ async def save(self, **kwargs: Any) -> None:  # type: ignore[override]
 
 
 class Question(Post):
-    if TYPE_CHECKING:
-        _id: Optional[int]
-
-    tags: M[List[str]]  # .tags will return empty list if not present
-    title: M[str] = mapped_field(Text(fields={"keyword": Keyword()}))
+    tags: List[str] = mapped_field(
+        default_factory=list
+    )  # .tags will return empty list if not present
+    title: str = mapped_field(Text(fields={"keyword": Keyword()}), default="")
 
     @classmethod
     def _matches(cls, hit: Dict[str, Any]) -> bool:
@@ -162,7 +168,6 @@ async def add_answer(
             created=created,
             body=body,
             is_accepted=accepted,
-            comments=[],
         )
         if commit:
             await answer.save()
@@ -192,11 +197,7 @@ async def save(self, **kwargs: Any) -> None:  # type: ignore[override]
 
 
 class Answer(Post):
-    if TYPE_CHECKING:
-        _routing: str
-        _index: AsyncIndex
-
-    is_accepted: M[bool]
+    is_accepted: bool = mapped_field(default=False)
 
     @classmethod
     def _matches(cls, hit: Dict[str, Any]) -> bool:
@@ -263,9 +264,6 @@ async def main() -> Answer:
         body="""
         I want to use elasticsearch, how do I do it from Python?
         """,
-        created=datetime.now(),
-        question_answer={},
-        comments=[],
     )
     await question.save()
     answer = await question.add_answer(honza, "Just use `elasticsearch-py`!")

examples/async/percolate.py

@@ -23,7 +23,6 @@
     AsyncDocument,
     AsyncSearch,
     Keyword,
-    M,
     Percolator,
     Q,
     Query,
@@ -40,10 +39,8 @@ class BlogPost(AsyncDocument):
     if TYPE_CHECKING:
         _id: int
 
-    content: M[Optional[str]]
-    tags: M[List[str]] = mapped_field(
-        Keyword(multi=True, required=False), default_factory=list
-    )
+    content: Optional[str]
+    tags: List[str] = mapped_field(Keyword(), default_factory=list)
 
     class Index:
         name = "test-blogpost"
@@ -78,12 +75,12 @@ class PercolatorDoc(AsyncDocument):
     # relevant fields from BlogPost must be also present here for the queries
     # to be able to use them. Another option would be to use document
     # inheritance but save() would have to be reset to normal behavior.
-    content: M[Optional[str]]
+    content: Optional[str]
 
     # the percolator query to be run against the doc
-    query: M[Query] = mapped_field(Percolator())
+    query: Query = mapped_field(Percolator())
     # list of tags to append to a document
-    tags: M[List[str]] = mapped_field(Keyword(multi=True))
+    tags: List[str] = mapped_field(Keyword(multi=True))
 
     class Index:
         name = "test-percolator"

examples/async/sparse_vectors.py

@@ -64,7 +64,7 @@
 import json
 import os
 from datetime import datetime
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 from urllib.request import urlopen
 
 import nltk  # type: ignore
@@ -75,7 +75,6 @@
     AsyncSearch,
     InnerDoc,
     Keyword,
-    M,
     Q,
     SparseVector,
     async_connections,
@@ -89,25 +88,25 @@
 
 
 class Passage(InnerDoc):
-    content: M[Optional[str]]
-    embedding: M[Dict[str, float]] = mapped_field(SparseVector(), init=False)
+    content: Optional[str]
+    embedding: Dict[str, float] = mapped_field(SparseVector(), init=False)
 
 
 class WorkplaceDoc(AsyncDocument):
     class Index:
         name = "workplace_documents_sparse"
         settings = {"default_pipeline": "elser_ingest_pipeline"}
 
-    name: M[Optional[str]]
-    summary: M[Optional[str]]
-    content: M[Optional[str]]
-    created: M[Optional[str]]
-    updated: M[Optional[datetime]]
-    url: M[Optional[str]] = mapped_field(Keyword())
-    category: M[Optional[str]] = mapped_field(Keyword())
-    passages: M[List[Passage]] = mapped_field(default=[])
+    name: str
+    summary: str
+    content: str
+    created: datetime
+    updated: Optional[datetime]
+    url: str = mapped_field(Keyword())
+    category: str = mapped_field(Keyword())
+    passages: List[Passage] = mapped_field(default=[])
 
-    _model = None
+    _model: Any = None
 
     def clean(self) -> None:
         # split the content into sentences

examples/async/vectors.py

@@ -48,7 +48,7 @@
 import json
 import os
 from datetime import datetime
-from typing import List, Optional, cast
+from typing import Any, List, Optional, cast
 from urllib.request import urlopen
 
 import nltk  # type: ignore
@@ -72,32 +72,34 @@
 # initialize sentence tokenizer
 nltk.download("punkt", quiet=True)
 
+# this will be the embedding model
+embedding_model: Any = None
+
 
 class Passage(InnerDoc):
-    content: M[str]
-    embedding: M[List[float]] = mapped_field(DenseVector())
+    content: str
+    embedding: List[float] = mapped_field(DenseVector())
 
 
 class WorkplaceDoc(AsyncDocument):
     class Index:
         name = "workplace_documents"
 
-    name: M[str]
-    summary: M[str]
-    content: M[str]
-    created: M[datetime]
-    updated: M[Optional[datetime]]
-    url: M[str] = mapped_field(Keyword(required=True))
-    category: M[str] = mapped_field(Keyword(required=True))
+    name: str
+    summary: str
+    content: str
+    created: datetime
+    updated: Optional[datetime]
+    url: str = mapped_field(Keyword(required=True))
+    category: str = mapped_field(Keyword(required=True))
     passages: M[List[Passage]] = mapped_field(default=[])
 
-    _model = None
-
     @classmethod
     def get_embedding(cls, input: str) -> List[float]:
-        if cls._model is None:
-            cls._model = SentenceTransformer(MODEL_NAME)
-        return cast(List[float], list(cls._model.encode(input)))
+        global embedding_model
+        if embedding_model is None:
+            embedding_model = SentenceTransformer(MODEL_NAME)
+        return cast(List[float], list(embedding_model.encode(input)))
 
     def clean(self) -> None:
         # split the content into sentences

examples/completion.py

@@ -35,7 +35,6 @@
     Document,
     Keyword,
     Long,
-    M,
     Text,
     analyzer,
     connections,
@@ -53,15 +52,15 @@
 
 
 class Person(Document):
-    name: M[str] = mapped_field(Text(fields={"keyword": Keyword()}))
-    popularity: M[int] = mapped_field(Long())
+    if TYPE_CHECKING:
+        _id: Optional[int] = mapped_field(default=None)
+
+    name: str = mapped_field(Text(fields={"keyword": Keyword()}), default="")
+    popularity: int = mapped_field(Long(), default=0)
 
     # completion field with a custom analyzer
     suggest: Dict[str, Any] = mapped_field(Completion(analyzer=ascii_fold), init=False)
 
-    if TYPE_CHECKING:
-        _id: Optional[int] = mapped_field(default=None)
-
     def clean(self) -> None:
         """
         Automatically construct the suggestion input and weight by taking all