🎨 Export developer API

esxport/__init__.py

@@ -1,3 +1,14 @@
 """EsXport CLi."""
+from .click_opt.cli_options import CliOptions
+from .elastic import ElasticsearchClient
+from .esxport import EsXport
 
 __version__ = "8.10.0"
+
+
+__all__ = [
+    # Core
+    "CliOptions",
+    "ElasticsearchClient",
+    "EsXport",
+]

esxport/click_opt/cli_options.py

@@ -8,6 +8,8 @@
 import urllib3
 from elastic_transport import SecurityWarning
 
+from esxport.constant import default_config_fields
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -24,21 +26,21 @@ def __init__(
     ) -> None:
         self.query: dict[str, Any] = myclass_kwargs["query"]
         self.output_file = myclass_kwargs["output_file"]
-        self.url = myclass_kwargs["url"]
-        self.user = myclass_kwargs["user"]
+        self.url = myclass_kwargs.get("url", default_config_fields["url"])
+        self.user = myclass_kwargs.get("user", default_config_fields["user"])
         self.password = myclass_kwargs["password"]
-        self.index_prefixes = myclass_kwargs["index_prefixes"]
-        self.fields: list[str] = list(myclass_kwargs["fields"])
-        self.sort: list[dict[str, str]] = myclass_kwargs["sort"]
-        self.delimiter = myclass_kwargs["delimiter"]
-        self.max_results = int(myclass_kwargs["max_results"])
-        self.scroll_size = int(myclass_kwargs["scroll_size"])
-        self.meta_fields: list[str] = list(myclass_kwargs["meta_fields"])
-        self.verify_certs: bool = myclass_kwargs["verify_certs"]
-        self.ca_certs = myclass_kwargs["ca_certs"]
-        self.client_cert = myclass_kwargs["ca_certs"]
-        self.client_key = myclass_kwargs["ca_certs"]
-        self.debug: bool = myclass_kwargs["debug"]
+        self.index_prefixes = myclass_kwargs.get("index_prefixes", default_config_fields["index_prefixes"])
+        self.fields: list[str] = list(myclass_kwargs.get("fields", default_config_fields["fields"]))
+        self.sort: list[dict[str, str]] = myclass_kwargs.get("sort", default_config_fields["sort"])
+        self.delimiter = myclass_kwargs.get("delimiter", default_config_fields["delimiter"])
+        self.max_results = int(myclass_kwargs.get("max_results", default_config_fields["max_results"]))
+        self.scroll_size = int(myclass_kwargs.get("scroll_size", default_config_fields["scroll_size"]))
+        self.meta_fields: list[str] = list(myclass_kwargs.get("meta_fields", default_config_fields["meta_fields"]))
+        self.verify_certs: bool = myclass_kwargs.get("verify_certs", default_config_fields["verify_certs"])
+        self.ca_certs = myclass_kwargs.get("ca_certs", default_config_fields["ca_certs"])
+        self.client_cert = myclass_kwargs.get("ca_certs", default_config_fields["client_cert"])
+        self.client_key = myclass_kwargs.get("ca_certs", default_config_fields["client_key"])
+        self.debug: bool = myclass_kwargs.get("debug", default_config_fields["debug"])
         self.format: str = "csv"
 
     def __str__(self: Self) -> str:

esxport/click_opt/click_custom.py

@@ -52,6 +52,8 @@ class Json(ParamType):
     def convert(self: Self, value: Any, param: Parameter | None, ctx: Context | None) -> dict[str, Any]:
         """Convert input to json."""
         try:
+            if isinstance(value, dict):
+                return value
             return json.loads(value)  # type: ignore[no-any-return]
 
         except json.JSONDecodeError as exc:

esxport/constant.py

@@ -4,3 +4,19 @@
 TIMES_TO_TRY = 3
 RETRY_DELAY = 60
 META_FIELDS = ["_id", "_index", "_score"]
+default_config_fields = {
+    "url": "https://localhost:9200",
+    "user": "elastic",
+    "index_prefixes": "",
+    "fields": ["_all"],
+    "sort": [],
+    "delimiter": ",",
+    "max_results": 10,
+    "scroll_size": 100,
+    "meta_fields": [],
+    "verify_certs": True,
+    "ca_certs": "",
+    "client_cert": "",
+    "client_key": "",
+    "debug": False,
+}

esxport/esxport.py

@@ -16,6 +16,8 @@
 from esxport.strings import index_not_found, meta_field_not_found, output_fields, sorting_by, using_indexes, using_query
 from esxport.writer import Writer
 
+from .click_opt.click_custom import Json
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -87,7 +89,7 @@ def _prepare_search_query(self: Self) -> None:
             "scroll": self.scroll_time,
             "size": self.opts.scroll_size,
             "terminate_after": self.opts.max_results,
-            "body": self.opts.query,
+            "query": Json().convert(self.opts.query["query"], None, None),
         }
         if self.opts.sort:
             self.search_args["sort"] = self.opts.sort

esxport/esxport_cli.py

@@ -7,13 +7,13 @@
 from click import Context, Parameter
 from click_params import URL
 
-from esxport import esxport
-from esxport.__init__ import __version__
-from esxport.click_opt.cli_options import CliOptions
-from esxport.click_opt.click_custom import JSON, sort
-from esxport.constant import META_FIELDS
-from esxport.elastic import ElasticsearchClient
-from esxport.strings import cli_version
+from .__init__ import __version__
+from .click_opt.cli_options import CliOptions
+from .click_opt.click_custom import JSON, sort
+from .constant import META_FIELDS, default_config_fields
+from .elastic import ElasticsearchClient
+from .esxport import EsXport
+from .strings import cli_version
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -48,14 +48,14 @@ def print_version(ctx: Context, _: Parameter, value: bool) -> None:  # noqa: FBT
     "--url",
     type=URL,
     required=False,
-    default="https://localhost:9200",
+    default=default_config_fields["url"],
     help="Elasticsearch host URL.",
 )
 @click.option(
     "-U",
     "--user",
     required=False,
-    default="elastic",
+    default=default_config_fields["user"],
     help="Elasticsearch basic authentication user.",
 )
 @click.password_option(
@@ -68,7 +68,7 @@ def print_version(ctx: Context, _: Parameter, value: bool) -> None:  # noqa: FBT
 @click.option(
     "-f",
     "--fields",
-    default=["_all"],
+    default=default_config_fields["fields"],
     multiple=True,
     help="List of _source fields to present be in output.",
 )
@@ -82,28 +82,28 @@ def print_version(ctx: Context, _: Parameter, value: bool) -> None:  # noqa: FBT
 @click.option(
     "-d",
     "--delimiter",
-    default=",",
+    default=default_config_fields["delimiter"],
     help="Delimiter to use in CSV file.",
 )
 @click.option(
     "-m",
     "--max-results",
-    default=10,
+    default=default_config_fields["max_results"],
     type=int,
     help="Maximum number of results to return.",
 )
 @click.option(
     "-s",
     "--scroll-size",
-    default=100,
+    default=default_config_fields["scroll_size"],
     type=int,
     help="Scroll size for each batch of results.",
 )
 @click.option(
     "-e",
     "--meta-fields",
     type=click.Choice(META_FIELDS),
-    default=[],
+    default=default_config_fields["meta_fields"],
     multiple=True,
     help="Add meta-fields in output.",
 )
@@ -138,7 +138,7 @@ def print_version(ctx: Context, _: Parameter, value: bool) -> None:  # noqa: FBT
 @click.option(
     "--debug",
     is_flag=True,
-    default=False,
+    default=default_config_fields["debug"],
     help="Debug mode on.",
 )
 def cli(  # noqa: PLR0913
@@ -164,7 +164,7 @@ def cli(  # noqa: PLR0913
     kwargs: dict[str, Any] = {k: v for k, v in locals().items() if k != "self"}
     cli_options = CliOptions(kwargs)
     client = ElasticsearchClient(cli_options)
-    es = esxport.EsXport(cli_options, client)
+    es = EsXport(cli_options, client)
     es.export()