handling of more data stream operations

Author: miguelgrinberg

elasticsearch/dsl/_async/index.py

@@ -227,6 +227,10 @@ async def create(
         Any additional keyword arguments will be passed to
         ``Elasticsearch.indices.create`` unchanged.
         """
+        if self._data_stream:
+            return await self._get_connection(using).indices.create_data_stream(
+                name=self._name, **kwargs
+            )
         return await self._get_connection(using).indices.create(
             index=self._name, body=self.to_dict(), **kwargs
         )
@@ -253,11 +257,14 @@ async def save(
         """
         if self._data_stream:
             template = self.as_composable_template(f"{self._name}-template", self._name)
-            return await template.save(using=using)
+            await template.save(using=using)
 
         if not await self.exists(using=using):
             return await self.create(using=using)
 
+        if self._data_stream:
+            return None  # the data stream's index template is already updated
+
         body = self.to_dict()
         settings = body.pop("settings", {})
         analysis = settings.pop("analysis", None)
@@ -388,6 +395,10 @@ async def delete(
         Any additional keyword arguments will be passed to
         ``Elasticsearch.indices.delete`` unchanged.
         """
+        if self._data_stream:
+            return await self._get_connection(using).indices.delete_data_stream(
+                name=self._name, **kwargs
+            )
         return await self._get_connection(using).indices.delete(
             index=self._name, **kwargs
         )

elasticsearch/dsl/_sync/index.py

@@ -215,6 +215,10 @@ def create(
         Any additional keyword arguments will be passed to
         ``Elasticsearch.indices.create`` unchanged.
         """
+        if self._data_stream:
+            return self._get_connection(using).indices.create_data_stream(
+                name=self._name, **kwargs
+            )
         return self._get_connection(using).indices.create(
             index=self._name, body=self.to_dict(), **kwargs
         )
@@ -241,11 +245,14 @@ def save(
         """
         if self._data_stream:
             template = self.as_composable_template(f"{self._name}-template", self._name)
-            return template.save(using=using)
+            template.save(using=using)
 
         if not self.exists(using=using):
             return self.create(using=using)
 
+        if self._data_stream:
+            return None  # the data stream's index template is already updated
+
         body = self.to_dict()
         settings = body.pop("settings", {})
         analysis = settings.pop("analysis", None)
@@ -366,6 +373,10 @@ def delete(
         Any additional keyword arguments will be passed to
         ``Elasticsearch.indices.delete`` unchanged.
         """
+        if self._data_stream:
+            return self._get_connection(using).indices.delete_data_stream(
+                name=self._name, **kwargs
+            )
         return self._get_connection(using).indices.delete(index=self._name, **kwargs)
 
     def exists(self, using: Optional[UsingType] = None, **kwargs: Any) -> bool:

examples/dsl/async/data_stream.py

@@ -29,7 +29,7 @@ async def main() -> None:
 
     # delete a previous instance of the data stream if one exists
     if await Log._index.exists():
-        await client.indices.delete_data_stream(name=Log._index._name)
+        await Log._index.delete()
 
     # create the data stream
     await Log.init()

examples/dsl/data_stream.py

@@ -26,7 +26,7 @@ def main() -> None:
 
     # delete a previous instance of the data stream if one exists
     if Log._index.exists():
-        client.indices.delete_data_stream(name=Log._index._name)
+        Log._index.delete()
 
     # create the data stream
     Log.init()