PY-175 Rename envs from NEPTUNE_FETCHER_* to NEPTUNE_QUERY_* (#412)

* Rename `NEPTUNE_FETCHER_QUERY_SIZE_LIMIT` env to `NEPTUNE_FETCHER_MAX_REQUEST_SIZE`

* PY-175 Rename envs from NEPTUNE_FETCHER_* to NEPTUNE_QUERY_*

* Sort env variables

* Bump default NEPTUNE_QUERY_MAX_WORKERS to 32 (from 10) (#413)

The workers don't do much CPU work, but rather wait for network IO and 32 was
shown to improve performance of fetching files. Other than a slight increase
in the memory usage, there's no known downside to increasing this number to 32.

Author: gabrys

docs/old/api_reference.md

@@ -424,7 +424,7 @@ Improves the performance of access to consecutive field values. Works only for s
 To speed up the fetching process, this method can use multithreading.
 To enable it, set the `use_threads` parameter to `True`.
 
-By default, the maximum number of workers is 10. You can change this number by setting the `NEPTUNE_FETCHER_MAX_WORKERS`
+By default, the maximum number of workers is 32. You can change this number by setting the `NEPTUNE_QUERY_MAX_WORKERS`
 environment variable.
 
 __Parameters__:

src/neptune_query/internal/composition/attributes.py

@@ -47,7 +47,7 @@ def fetch_attribute_definitions(
     run_identifiers: Optional[Iterable[identifiers.RunIdentifier]],
     attribute_filter: filters._BaseAttributeFilter,
     executor: Executor,
-    batch_size: int = env.NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
+    batch_size: int = env.NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
 ) -> Generator[util.Page[identifiers.AttributeDefinition], None, None]:
     pages_filters = _fetch_attribute_definitions(
         client, project_identifiers, run_identifiers, attribute_filter, batch_size, executor
@@ -66,7 +66,7 @@ def fetch_attribute_definition_aggregations(
     run_identifiers: Iterable[identifiers.RunIdentifier],
     attribute_filter: filters._BaseAttributeFilter,
     executor: Executor,
-    batch_size: int = env.NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
+    batch_size: int = env.NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
 ) -> Generator[
     tuple[util.Page[identifiers.AttributeDefinition], util.Page[AttributeDefinitionAggregation]], None, None
 ]:

src/neptune_query/internal/composition/concurrency.py

@@ -43,7 +43,7 @@
 
 
 def create_thread_pool_executor() -> Executor:
-    max_workers = env.NEPTUNE_FETCHER_MAX_WORKERS.get()
+    max_workers = env.NEPTUNE_QUERY_MAX_WORKERS.get()
     return ThreadPoolExecutor(max_workers=max_workers)
 
 

src/neptune_query/internal/env.py

@@ -22,21 +22,21 @@
 )
 
 __all__ = (
-    "NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS",
     "NEPTUNE_API_TOKEN",
-    "NEPTUNE_FETCHER_MAX_WORKERS",
+    "NEPTUNE_ENABLE_COLORS",
+    "NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS",
     "NEPTUNE_PROJECT",
+    "NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE",
+    "NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE",
+    "NEPTUNE_QUERY_FILES_MAX_CONCURRENCY",
+    "NEPTUNE_QUERY_FILES_TIMEOUT",
+    "NEPTUNE_QUERY_MAX_REQUEST_SIZE",
+    "NEPTUNE_QUERY_MAX_WORKERS",
+    "NEPTUNE_QUERY_RETRY_HARD_TIMEOUT",
+    "NEPTUNE_QUERY_RETRY_SOFT_TIMEOUT",
+    "NEPTUNE_QUERY_SERIES_BATCH_SIZE",
+    "NEPTUNE_QUERY_SYS_ATTRS_BATCH_SIZE",
     "NEPTUNE_VERIFY_SSL",
-    "NEPTUNE_FETCHER_RETRY_SOFT_TIMEOUT",
-    "NEPTUNE_FETCHER_RETRY_HARD_TIMEOUT",
-    "NEPTUNE_FETCHER_SYS_ATTRS_BATCH_SIZE",
-    "NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE",
-    "NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE",
-    "NEPTUNE_FETCHER_SERIES_BATCH_SIZE",
-    "NEPTUNE_FETCHER_QUERY_SIZE_LIMIT",
-    "NEPTUNE_FETCHER_FILES_MAX_CONCURRENCY",
-    "NEPTUNE_FETCHER_FILES_TIMEOUT",
-    "NEPTUNE_ENABLE_COLORS",
 )
 
 T = TypeVar("T")
@@ -79,28 +79,25 @@ def wrapped(value: str) -> Optional[T]:
     return wrapped
 
 
-NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS = EnvVariable[int]("NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS", int, 60)
 NEPTUNE_API_TOKEN = EnvVariable[str]("NEPTUNE_API_TOKEN", _map_str)
+NEPTUNE_ENABLE_COLORS = EnvVariable[bool]("NEPTUNE_ENABLE_COLORS", _map_bool, True)
+NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS = EnvVariable[int]("NEPTUNE_HTTP_REQUEST_TIMEOUT_SECONDS", int, 60)
 NEPTUNE_PROJECT = EnvVariable[str]("NEPTUNE_PROJECT", _map_str)
-NEPTUNE_VERIFY_SSL = EnvVariable[bool]("NEPTUNE_VERIFY_SSL", _map_bool, True)
-NEPTUNE_FETCHER_RETRY_SOFT_TIMEOUT = EnvVariable[Optional[int]](
-    "NEPTUNE_FETCHER_RETRY_SOFT_TIMEOUT", _lift_optional(int), 300
-)
-NEPTUNE_FETCHER_RETRY_HARD_TIMEOUT = EnvVariable[Optional[int]](
-    "NEPTUNE_FETCHER_RETRY_HARD_TIMEOUT", _lift_optional(int), 600
+NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE = EnvVariable[int](
+    "NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE", int, 10_000
 )
-NEPTUNE_FETCHER_MAX_WORKERS = EnvVariable[int]("NEPTUNE_FETCHER_MAX_WORKERS", int, 10)
-NEPTUNE_FETCHER_SYS_ATTRS_BATCH_SIZE = EnvVariable[int]("NEPTUNE_FETCHER_EXPERIMENT_SYS_ATTRS_BATCH_SIZE", int, 10_000)
-NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE = EnvVariable[int](
-    "NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE", int, 10_000
+NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE = EnvVariable[int]("NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE", int, 10_000)
+NEPTUNE_QUERY_FILES_BATCH_SIZE = EnvVariable[int]("NEPTUNE_QUERY_FILES_BATCH_SIZE", int, 200)
+NEPTUNE_QUERY_FILES_MAX_CONCURRENCY = EnvVariable[int]("NEPTUNE_QUERY_FILES_MAX_CONCURRENCY", int, 1)
+NEPTUNE_QUERY_FILES_TIMEOUT = EnvVariable[Optional[int]]("NEPTUNE_QUERY_FILES_TIMEOUT", _lift_optional(int), None)
+NEPTUNE_QUERY_MAX_REQUEST_SIZE = EnvVariable[int]("NEPTUNE_QUERY_MAX_REQUEST_SIZE", int, 220_000)
+NEPTUNE_QUERY_MAX_WORKERS = EnvVariable[int]("NEPTUNE_QUERY_MAX_WORKERS", int, 32)
+NEPTUNE_QUERY_RETRY_HARD_TIMEOUT = EnvVariable[Optional[int]](
+    "NEPTUNE_QUERY_RETRY_HARD_TIMEOUT", _lift_optional(int), 600
 )
-NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE = EnvVariable[int](
-    "NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE", int, 10_000
+NEPTUNE_QUERY_RETRY_SOFT_TIMEOUT = EnvVariable[Optional[int]](
+    "NEPTUNE_QUERY_RETRY_SOFT_TIMEOUT", _lift_optional(int), 300
 )
-NEPTUNE_FETCHER_SERIES_BATCH_SIZE = EnvVariable[int]("NEPTUNE_FETCHER_SERIES_BATCH_SIZE", int, 10_000)
-NEPTUNE_FETCHER_QUERY_SIZE_LIMIT = EnvVariable[int]("NEPTUNE_FETCHER_QUERY_SIZE_LIMIT", int, 220_000)
-NEPTUNE_FETCHER_FILES_BATCH_SIZE = EnvVariable[int]("NEPTUNE_FETCHER_FILES_BATCH_SIZE", int, 200)
-NEPTUNE_FETCHER_FILES_MAX_CONCURRENCY = EnvVariable[int]("NEPTUNE_FETCHER_FILES_MAX_CONCURRENCY", int, 1)
-NEPTUNE_FETCHER_FILES_TIMEOUT = EnvVariable[Optional[int]]("NEPTUNE_FETCHER_FILES_TIMEOUT", _lift_optional(int), None)
-
-NEPTUNE_ENABLE_COLORS = EnvVariable[bool]("NEPTUNE_ENABLE_COLORS", _map_bool, True)
+NEPTUNE_QUERY_SERIES_BATCH_SIZE = EnvVariable[int]("NEPTUNE_QUERY_SERIES_BATCH_SIZE", int, 10_000)
+NEPTUNE_QUERY_SYS_ATTRS_BATCH_SIZE = EnvVariable[int]("NEPTUNE_QUERY_EXPERIMENT_SYS_ATTRS_BATCH_SIZE", int, 10_000)
+NEPTUNE_VERIFY_SSL = EnvVariable[bool]("NEPTUNE_VERIFY_SSL", _map_bool, True)

src/neptune_query/internal/retrieval/attribute_definitions.py

@@ -58,7 +58,7 @@ def fetch_attribute_definitions_single_filter(
     project_identifiers: Iterable[identifiers.ProjectIdentifier],
     run_identifiers: Optional[Iterable[identifiers.RunIdentifier]],
     attribute_filter: filters._AttributeFilter,
-    batch_size: int = env.NEPTUNE_FETCHER_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
+    batch_size: int = env.NEPTUNE_QUERY_ATTRIBUTE_DEFINITIONS_BATCH_SIZE.get(),
 ) -> Generator[util.Page[identifiers.AttributeDefinition], None, None]:
     params: dict[str, Any] = {
         "projectIdentifiers": list(project_identifiers),

src/neptune_query/internal/retrieval/attribute_values.py

@@ -54,7 +54,7 @@ def fetch_attribute_values(
     project_identifier: identifiers.ProjectIdentifier,
     run_identifiers: Iterable[identifiers.RunIdentifier],
     attribute_definitions: Iterable[identifiers.AttributeDefinition],
-    batch_size: int = env.NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE.get(),
+    batch_size: int = env.NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE.get(),
 ) -> Generator[util.Page[AttributeValue], None, None]:
     attribute_definitions_set: set[identifiers.AttributeDefinition] = set(attribute_definitions)
     experiments = [str(e) for e in run_identifiers]

src/neptune_query/internal/retrieval/files.py

@@ -118,8 +118,8 @@ class DownloadResult:
 def download_file(
     signed_file: SignedFile,
     target_path: pathlib.Path,
-    max_concurrency: int = env.NEPTUNE_FETCHER_FILES_MAX_CONCURRENCY.get(),
-    timeout: Optional[int] = env.NEPTUNE_FETCHER_FILES_TIMEOUT.get(),
+    max_concurrency: int = env.NEPTUNE_QUERY_FILES_MAX_CONCURRENCY.get(),
+    timeout: Optional[int] = env.NEPTUNE_QUERY_FILES_TIMEOUT.get(),
 ) -> DownloadResult:
     target_path.parent.mkdir(parents=True, exist_ok=True)
 
@@ -136,8 +136,8 @@ def download_file(
 def _download_file_azure(
     signed_file: SignedFile,
     target_path: pathlib.Path,
-    max_concurrency: int = env.NEPTUNE_FETCHER_FILES_MAX_CONCURRENCY.get(),
-    timeout: Optional[int] = env.NEPTUNE_FETCHER_FILES_TIMEOUT.get(),
+    max_concurrency: int = env.NEPTUNE_QUERY_FILES_MAX_CONCURRENCY.get(),
+    timeout: Optional[int] = env.NEPTUNE_QUERY_FILES_TIMEOUT.get(),
 ) -> DownloadResult:
     try:
         blob_client = AzureBlobClient.from_blob_url(signed_file.url)
@@ -160,7 +160,7 @@ def _download_file_azure(
 def _download_file_requests(
     signed_file: SignedFile,
     target_path: pathlib.Path,
-    timeout: Optional[int] = env.NEPTUNE_FETCHER_FILES_TIMEOUT.get(),
+    timeout: Optional[int] = env.NEPTUNE_QUERY_FILES_TIMEOUT.get(),
 ) -> DownloadResult:
     try:
         response = requests.get(signed_file.url, stream=True, timeout=timeout)

src/neptune_query/internal/retrieval/retry.py

@@ -44,8 +44,8 @@
 def handle_errors_default(func: Callable[T, Response[R]]) -> Callable[T, Response[R]]:
     return retry_backoff(
         max_tries=None,
-        soft_max_time=env.NEPTUNE_FETCHER_RETRY_SOFT_TIMEOUT.get(),
-        hard_max_time=env.NEPTUNE_FETCHER_RETRY_HARD_TIMEOUT.get(),
+        soft_max_time=env.NEPTUNE_QUERY_RETRY_SOFT_TIMEOUT.get(),
+        hard_max_time=env.NEPTUNE_QUERY_RETRY_HARD_TIMEOUT.get(),
         backoff_strategy=exponential_backoff(jitter="full"),
     )(handle_api_errors(func))
 

src/neptune_query/internal/retrieval/search.py

@@ -130,7 +130,7 @@ def __call__(
         sort_by: _Attribute = _Attribute("sys/creation_time", type="datetime"),
         sort_direction: Literal["asc", "desc"] = "desc",
         limit: Optional[int] = None,
-        batch_size: int = env.NEPTUNE_FETCHER_SYS_ATTRS_BATCH_SIZE.get(),
+        batch_size: int = env.NEPTUNE_QUERY_SYS_ATTRS_BATCH_SIZE.get(),
         container_type: ContainerType = ContainerType.EXPERIMENT,
     ) -> Generator[util.Page[T], None, None]:
         ...
@@ -148,7 +148,7 @@ def fetch_sys_attrs(
         sort_by: _Attribute = _Attribute("sys/creation_time", type="datetime"),
         sort_direction: Literal["asc", "desc"] = "desc",
         limit: Optional[int] = None,
-        batch_size: int = env.NEPTUNE_FETCHER_SYS_ATTRS_BATCH_SIZE.get(),
+        batch_size: int = env.NEPTUNE_QUERY_SYS_ATTRS_BATCH_SIZE.get(),
         container_type: ContainerType = default_container_type,
     ) -> Generator[util.Page[T], None, None]:
         params: dict[str, Any] = {

src/neptune_query/internal/retrieval/split.py

@@ -49,10 +49,10 @@ def split_sys_ids(
     sys_ids: list[identifiers.SysId],
 ) -> Generator[list[identifiers.SysId]]:
     """
-    Splits a sequence of sys ids into batches of size at most `NEPTUNE_FETCHER_QUERY_SIZE_LIMIT`.
+    Splits a sequence of sys ids into batches of size at most `NEPTUNE_QUERY_MAX_REQUEST_SIZE`.
     Use before fetching attribute definitions.
     """
-    query_size_limit = env.NEPTUNE_FETCHER_QUERY_SIZE_LIMIT.get()
+    query_size_limit = env.NEPTUNE_QUERY_MAX_REQUEST_SIZE.get()
     identifier_num_limit = max(query_size_limit // _sys_id_size(), 1)
 
     identifier_num = len(sys_ids)
@@ -72,14 +72,14 @@ def split_sys_ids_attributes(
 ) -> Generator[tuple[list[identifiers.SysId], list[identifiers.AttributeDefinition]]]:
     """
     Splits a pair of sys ids and attribute_definitions into batches that:
-    When their length is added it is of size at most `NEPTUNE_FETCHER_QUERY_SIZE_LIMIT`.
-    When their item count is multiplied, it is at most `NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE`.
+    When their length is added it is of size at most `NEPTUNE_QUERY_MAX_REQUEST_SIZE`.
+    When their item count is multiplied, it is at most `NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE`.
 
     It's intended for use before fetching attribute values and assumes that the sys_ids and attribute_definitions
     will be sent to the server in a single request and the response will contain data for their cartesian product.
     """
-    query_size_limit = env.NEPTUNE_FETCHER_QUERY_SIZE_LIMIT.get()
-    attribute_values_batch_size = env.NEPTUNE_FETCHER_ATTRIBUTE_VALUES_BATCH_SIZE.get()
+    query_size_limit = env.NEPTUNE_QUERY_MAX_REQUEST_SIZE.get()
+    attribute_values_batch_size = env.NEPTUNE_QUERY_ATTRIBUTE_VALUES_BATCH_SIZE.get()
 
     if not attribute_definitions:
         return
@@ -141,13 +141,13 @@ def _split_attribute_definitions(
 def split_series_attributes(items: Iterable[RunAttributeDefinition]) -> Generator[list[RunAttributeDefinition]]:
     """
     Splits a list of classes containing an attribute_definition into batches so that:
-    When the lengths of attribute paths are added, the total length is at most `NEPTUNE_FETCHER_QUERY_SIZE_LIMIT`.
-    Item count is at most `NEPTUNE_FETCHER_SERIES_BATCH_SIZE`.
+    When the lengths of attribute paths are added, the total length is at most `NEPTUNE_QUERY_MAX_REQUEST_SIZE`.
+    Item count is at most `NEPTUNE_QUERY_SERIES_BATCH_SIZE`.
 
     Intended for use before fetching (string, float) series.
     """
-    query_size_limit = env.NEPTUNE_FETCHER_QUERY_SIZE_LIMIT.get()
-    batch_size_limit = env.NEPTUNE_FETCHER_SERIES_BATCH_SIZE.get()
+    query_size_limit = env.NEPTUNE_QUERY_MAX_REQUEST_SIZE.get()
+    batch_size_limit = env.NEPTUNE_QUERY_SERIES_BATCH_SIZE.get()
 
     if not items:
         return
@@ -172,7 +172,7 @@ def _ceil_div(a: int, b: int) -> int:
 
 
 def split_files(items: list[T]) -> Generator[list[T], None, None]:
-    batch_size = env.NEPTUNE_FETCHER_FILES_BATCH_SIZE.get()
+    batch_size = env.NEPTUNE_QUERY_FILES_BATCH_SIZE.get()
 
     for i in range(0, len(items), batch_size):
         yield items[i : i + batch_size]