from enum import Enum
from numbers import Number
from typing import (
Any,
Callable,
Dict,
Iterator,
List,
MutableMapping,
Optional,
Tuple,
TypeAlias,
cast,
)
from warnings import warn
from multidict import CIMultiDictProxy, MultiDict
Json: TypeAlias = Dict[str, Any]
"""Type definition for request/response body"""
Jsons: TypeAlias = List[Json]
"""Type definition for a list of JSON objects"""
RequestHeaders: TypeAlias = MutableMapping[str, str] | MultiDict[str]
"""Type definition for request HTTP headers"""
ResponseHeaders: TypeAlias = (
MutableMapping[str, str] | MultiDict[str] | CIMultiDictProxy[str]
)
"""Type definition for response HTTP headers"""
Params: TypeAlias = MutableMapping[str, bool | int | str | float]
"""Type definition for URL (query) parameters"""
Formatter: TypeAlias = Callable[[Json], Json]
"""Type definition for a JSON formatter"""
[docs]
class CollectionType(Enum):
"""Collection types."""
DOCUMENT = 2
EDGE = 3
[docs]
@staticmethod
def from_int(value: int) -> "CollectionType":
"""Return a collection type from its integer value.
Args:
value (int): Collection type integer value.
Returns:
CollectionType: Collection type.
"""
if value == 2:
return CollectionType.DOCUMENT
elif value == 3:
return CollectionType.EDGE
else:
raise ValueError(f"Invalid collection type value: {value}")
[docs]
@staticmethod
def from_str(value: str) -> "CollectionType":
"""Return a collection type from its string value.
Args:
value (str): Collection type string value.
Returns:
CollectionType: Collection type.
"""
value = value.lower()
if value == "document":
return CollectionType.DOCUMENT
elif value == "edge":
return CollectionType.EDGE
else:
raise ValueError(f"Invalid collection type value: {value}")
def __str__(self) -> str:
return self.name.lower()
def __repr__(self) -> str:
return self.name.lower()
[docs]
class CollectionStatus(Enum):
"""Collection status."""
NEW = 1
UNLOADED = 2
LOADED = 3
UNLOADING = 4
DELETED = 5
LOADING = 6
[docs]
@staticmethod
def from_int(value: int) -> "CollectionStatus":
"""Return a collection status from its integer value.
Args:
value (int): Collection status integer value.
Returns:
CollectionStatus: Collection status.
"""
if value == 1:
return CollectionStatus.NEW
elif value == 2:
return CollectionStatus.UNLOADED
elif value == 3:
return CollectionStatus.LOADED
elif value == 4:
return CollectionStatus.UNLOADING
elif value == 5:
return CollectionStatus.DELETED
elif value == 6:
return CollectionStatus.LOADING
else:
raise ValueError(f"Invalid collection status value: {value}")
def __str__(self) -> str:
return self.name.lower()
def __repr__(self) -> str:
return self.name.lower()
[docs]
class JsonWrapper:
"""Wrapper over server request/response objects."""
def __init__(self, data: Json) -> None:
self._data = data
for excluded in ("code", "error"):
if excluded in self._data:
self._data.pop(excluded)
def __getitem__(self, key: str) -> Any:
return self._data[key]
def __setitem__(self, key: str, value: Any) -> None:
self._data[key] = value
def __delitem__(self, key: str) -> None:
del self._data[key]
def __iter__(self) -> Iterator[str]:
return iter(self._data)
def __len__(self) -> int:
return len(self._data)
def __contains__(self, item: str) -> bool:
return item in self._data
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._data})"
def __str__(self) -> str:
return str(self._data)
def __eq__(self, other: object) -> bool:
return self._data == other
[docs]
def get(self, key: str, default: Optional[Any] = None) -> Any:
"""Return the value for key if key is in the dictionary, else default."""
return self._data.get(key, default)
[docs]
def items(self) -> Iterator[Tuple[str, Any]]:
"""Return an iterator over the dictionary’s key-value pairs."""
return iter(self._data.items())
[docs]
def keys(self) -> Iterator[str]:
"""Return an iterator over the dictionary’s keys."""
return iter(self._data.keys())
[docs]
def values(self) -> Iterator[Any]:
"""Return an iterator over the dictionary’s values."""
return iter(self._data.values())
[docs]
def to_dict(self) -> Json:
"""Return the dictionary."""
return self._data
@staticmethod
def _strip_result(data: Json) -> Json:
"""Keep only the `result` key from a dict. Useful when parsing responses."""
return data["result"] # type: ignore[no-any-return]
[docs]
class KeyOptions(JsonWrapper):
"""Additional options for key generation, used on collections.
Args:
allow_user_keys (bool): If set to `True`, then you are allowed to supply own
key values in the `_key` attribute of documents. If set to `False`, then
the key generator is solely responsible for generating keys and an error
is raised if you supply own key values in the `_key` attribute of
documents.
generator_type (str): Specifies the type of the key generator. The currently
available generators are "traditional", "autoincrement", "uuid" and
"padded".
increment (int | None): The increment value for the "autoincrement" key
generator. Not allowed for other key generator types.
offset (int | None): The initial offset value for the "autoincrement" key
generator. Not allowed for other key generator types.
data (dict | None): Key options. If this parameter is specified, the
other parameters are ignored.
Example:
.. code-block:: json
{
"type": "autoincrement",
"increment": 5,
"allowUserKeys": true
}
References:
- `create-a-collection <https://docs.arango.ai/arangodb/stable/develop/http-api/collections/#create-a-collection_body_keyOptions>`__
""" # noqa: E501
def __init__(
self,
allow_user_keys: bool = True,
generator_type: str = "traditional",
increment: Optional[int] = None,
offset: Optional[int] = None,
data: Optional[Json] = None,
) -> None:
if data is None:
data: Json = { # type: ignore[no-redef]
"allowUserKeys": allow_user_keys,
"type": generator_type,
}
if increment is not None:
data["increment"] = increment # type: ignore[index]
if offset is not None:
data["offset"] = offset # type: ignore[index]
super().__init__(cast(Json, data))
[docs]
def validate(self) -> None:
"""Validate key options."""
if "type" not in self:
raise ValueError('"type" value is required for key options')
if "allowUserKeys" not in self:
raise ValueError('"allowUserKeys" value is required for key options')
allowed_types = {"autoincrement", "uuid", "padded", "traditional"}
if self["type"] not in allowed_types:
raise ValueError(
f"Invalid key generator type '{self['type']}', "
f"expected one of {allowed_types}"
)
if self.get("increment") is not None and self["type"] != "autoincrement":
raise ValueError(
'"increment" value is only allowed for "autoincrement" ' "key generator"
)
if self.get("offset") is not None and self["type"] != "autoincrement":
raise ValueError(
'"offset" value is only allowed for "autoincrement" ' "key generator"
)
[docs]
class CollectionInfo(JsonWrapper):
"""Collection information.
Example:
.. code-block:: json
{
"id" : "151",
"name" : "animals",
"status" : 3,
"type" : 2,
"isSystem" : false,
"globallyUniqueId" : "hDA74058C1843/151"
}
References:
- `get-the-collection-information <https://docs.arango.ai/arangodb/stable/develop/http-api/collections/#get-the-collection-information>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def globally_unique_id(self) -> str:
""" "A unique identifier of the collection (internal property)."""
return cast(str, self._data["globallyUniqueId"])
@property
def is_system(self) -> bool:
"""Whether the collection is a system collection."""
return cast(bool, self._data["isSystem"])
@property
def name(self) -> str:
"""Return the name of the collection."""
return cast(str, self._data["name"])
@property
def status(self) -> CollectionStatus:
"""Return the status of the collection."""
return CollectionStatus.from_int(self._data["status"])
@property
def col_type(self) -> CollectionType:
"""Return the type of the collection."""
return CollectionType.from_int(self._data["type"])
[docs]
class UserInfo(JsonWrapper):
"""User information.
Args:
user (str): The name of the user.
password (str | None): The user password as a string. Note that user
password is not returned back by the server.
active (bool): `True` if user is active, `False` otherwise.
extra (dict | None): Additional user information. For internal use only.
Should not be set or modified by end users.
Example:
.. code-block:: json
{
"user": "john",
"password": "secret",
"active": true,
"extra": {}
}
References:
- `create-a-user <https://docs.arango.ai/arangodb/stable/develop/http-api/users/#create-a-user>`__
""" # noqa: E501
def __init__(
self,
user: str,
password: Optional[str] = None,
active: bool = True,
extra: Optional[Json] = None,
) -> None:
data: Json = {"user": user, "active": active}
if password is not None:
data["password"] = password
if extra is not None:
data["extra"] = extra
super().__init__(data)
@property
def user(self) -> str:
return self._data["user"] # type: ignore[no-any-return]
@property
def password(self) -> Optional[str]:
return self._data.get("password")
@property
def active(self) -> bool:
return self._data["active"] # type: ignore[no-any-return]
@property
def extra(self) -> Optional[Json]:
return self._data.get("extra")
[docs]
class DatabaseProperties(JsonWrapper):
"""Properties of the database.
References:
- `get-information-about-the-current-database <https://docs.arango.ai/arangodb/stable/develop/http-api/databases/#get-information-about-the-current-database>`__
""" # noqa: E501
def __init__(self, data: Json, strip_result: bool = False) -> None:
super().__init__(self._strip_result(data) if strip_result else data)
@property
def name(self) -> str:
"""The name of the current database."""
return self._data["name"] # type: ignore[no-any-return]
@property
def id(self) -> str:
"""The id of the current database."""
return self._data["id"] # type: ignore[no-any-return]
@property
def path(self) -> Optional[str]:
"""The filesystem path of the current database. Removed in ArangoDB 4.0."""
return self._data.get("path")
@property
def is_system(self) -> bool:
"""Whether the database is the `_system` database."""
return self._data["isSystem"] # type: ignore[no-any-return]
@property
def sharding(self) -> Optional[str]:
"""The default sharding method for collections."""
return self._data.get("sharding")
@property
def replication_factor(self) -> Optional[int]:
"""The default replication factor for collections."""
return self._data.get("replicationFactor")
@property
def write_concern(self) -> Optional[int]:
"""The default write concern for collections."""
return self._data.get("writeConcern")
[docs]
class CollectionProperties(JsonWrapper):
"""Properties of a collection.
Example:
.. code-block:: json
{
"writeConcern" : 1,
"waitForSync" : true,
"usesRevisionsAsDocumentIds" : true,
"syncByRevision" : true,
"statusString" : "loaded",
"id" : "68452",
"isSmartChild" : false,
"schema" : null,
"name" : "products",
"type" : 2,
"status" : 3,
"cacheEnabled" : false,
"isSystem" : false,
"internalValidatorType" : 0,
"globallyUniqueId" : "hDA74058C1843/68452",
"keyOptions" : {
"allowUserKeys" : true,
"type" : "traditional",
"lastValue" : 0
},
"computedValues" : null,
"objectId" : "68453"
}
References:
- `get-the-properties-of-a-collection <https://docs.arango.ai/arangodb/stable/develop/http-api/collections/#get-the-properties-of-a-collection>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def write_concern(self) -> Optional[int]:
return self._data.get("writeConcern")
@property
def wait_for_sync(self) -> Optional[bool]:
return self._data.get("waitForSync")
@property
def use_revisions_as_document_ids(self) -> Optional[bool]:
return self._data.get("usesRevisionsAsDocumentIds")
@property
def sync_by_revision(self) -> Optional[bool]:
return self._data.get("syncByRevision")
@property
def status_string(self) -> Optional[str]:
m = "statusString attribute has been removed in ArangoDB 4.0"
warn(m, DeprecationWarning, stacklevel=2)
return self._data.get("statusString")
@property
def id(self) -> str:
return self._data["id"] # type: ignore[no-any-return]
@property
def is_smart_child(self) -> bool:
return self._data["isSmartChild"] # type: ignore[no-any-return]
@property
def schema(self) -> Optional[Json]:
return self._data.get("schema")
@property
def name(self) -> str:
return self._data["name"] # type: ignore[no-any-return]
@property
def type(self) -> CollectionType:
return CollectionType.from_int(self._data["type"])
@property
def status(self) -> CollectionStatus:
m = "status attribute has been removed in ArangoDB 4.0"
warn(m, DeprecationWarning, stacklevel=2)
return CollectionStatus.from_int(self._data["status"])
@property
def cache_enabled(self) -> Optional[bool]:
return self._data.get("cacheEnabled")
@property
def is_system(self) -> bool:
return self._data["isSystem"] # type: ignore[no-any-return]
@property
def internal_validator_type(self) -> Optional[int]:
return self._data.get("internalValidatorType")
@property
def globally_unique_id(self) -> str:
return self._data["globallyUniqueId"] # type: ignore[no-any-return]
@property
def key_options(self) -> KeyOptions:
return KeyOptions(self._data["keyOptions"])
@property
def computed_values(self) -> Optional[Json]:
return self._data.get("computedValues")
@property
def supportsRBAC(self) -> Optional[bool]:
return self._data.get("supportsRBAC")
@property
def object_id(self) -> str:
return self._data["objectId"] # type: ignore[no-any-return]
[docs]
class CollectionStatistics(JsonWrapper):
"""Statistical information about the collection.
Example:
.. code-block:: json
{
"figures" : {
"indexes" : {
"count" : 1,
"size" : 1234
},
"documentsSize" : 5601,
"cacheInUse" : false,
"cacheSize" : 0,
"cacheUsage" : 0,
"engine" : {
"documents" : 1,
"indexes" : [
{
"type" : "primary",
"id" : 0,
"count" : 1
}
]
}
},
"writeConcern" : 1,
"waitForSync" : false,
"usesRevisionsAsDocumentIds" : true,
"syncByRevision" : true,
"statusString" : "loaded",
"id" : "69123",
"isSmartChild" : false,
"schema" : null,
"name" : "products",
"type" : 2,
"status" : 3,
"count" : 1,
"cacheEnabled" : false,
"isSystem" : false,
"internalValidatorType" : 0,
"globallyUniqueId" : "hB7C02EE43DCE/69123",
"keyOptions" : {
"allowUserKeys" : true,
"type" : "traditional",
"lastValue" : 69129
},
"computedValues" : null,
"objectId" : "69124"
}
References:
- `get-the-collection-statistics <https://docs.arango.ai/arangodb/stable/develop/http-api/collections/#get-the-collection-statistics>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def figures(self) -> Json:
return cast(Json, self._data.get("figures"))
@property
def write_concern(self) -> Optional[int]:
return self._data.get("writeConcern")
@property
def wait_for_sync(self) -> Optional[bool]:
return self._data.get("waitForSync")
@property
def use_revisions_as_document_ids(self) -> Optional[bool]:
return self._data.get("usesRevisionsAsDocumentIds")
@property
def sync_by_revision(self) -> Optional[bool]:
return self._data.get("syncByRevision")
@property
def status_string(self) -> Optional[str]:
return self._data.get("statusString")
@property
def id(self) -> str:
return self._data["id"] # type: ignore[no-any-return]
@property
def is_smart_child(self) -> bool:
return self._data["isSmartChild"] # type: ignore[no-any-return]
@property
def schema(self) -> Optional[Json]:
return self._data.get("schema")
@property
def name(self) -> str:
return self._data["name"] # type: ignore[no-any-return]
@property
def type(self) -> CollectionType:
return CollectionType.from_int(self._data["type"])
@property
def status(self) -> CollectionStatus:
return CollectionStatus.from_int(self._data["status"])
@property
def count(self) -> int:
return self._data["count"] # type: ignore[no-any-return]
@property
def cache_enabled(self) -> Optional[bool]:
return self._data.get("cacheEnabled")
@property
def is_system(self) -> bool:
return self._data["isSystem"] # type: ignore[no-any-return]
@property
def internal_validator_type(self) -> Optional[int]:
return self._data.get("internalValidatorType")
@property
def globally_unique_id(self) -> str:
return self._data["globallyUniqueId"] # type: ignore[no-any-return]
@property
def key_options(self) -> KeyOptions:
return KeyOptions(self._data["keyOptions"])
@property
def computed_values(self) -> Optional[Json]:
return self._data.get("computedValues")
@property
def object_id(self) -> str:
return self._data["objectId"] # type: ignore[no-any-return]
[docs]
class IndexProperties(JsonWrapper):
"""Properties of an index.
Example:
.. code-block:: json
{
"fields" : [
"_key"
],
"id" : "products/0",
"name" : "primary",
"selectivityEstimate" : 1,
"sparse" : false,
"type" : "primary",
"unique" : true,
}
References:
- `get-an-index <https://docs.arango.ai/arangodb/stable/develop/http-api/indexes/#get-an-index>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def id(self) -> str:
return self._data["id"] # type: ignore[no-any-return]
@property
def numeric_id(self) -> int:
return int(self._data["id"].split("/", 1)[-1])
@property
def type(self) -> str:
return self._data["type"] # type: ignore[no-any-return]
@property
def fields(self) -> Json | List[str]:
return self._data["fields"] # type: ignore[no-any-return]
@property
def name(self) -> Optional[str]:
return self._data.get("name")
@property
def deduplicate(self) -> Optional[bool]:
return self._data.get("deduplicate")
@property
def sparse(self) -> Optional[bool]:
return self._data.get("sparse")
@property
def unique(self) -> Optional[bool]:
return self._data.get("unique")
@property
def geo_json(self) -> Optional[bool]:
return self._data.get("geoJson")
@property
def selectivity_estimate(self) -> Optional[float]:
return self._data.get("selectivityEstimate")
@property
def is_newly_created(self) -> Optional[bool]:
return self._data.get("isNewlyCreated")
@property
def expire_after(self) -> Optional[int]:
return self._data.get("expireAfter")
@property
def in_background(self) -> Optional[bool]:
return self._data.get("inBackground")
@property
def max_num_cover_cells(self) -> Optional[int]:
return self._data.get("maxNumCoverCells")
@property
def cache_enabled(self) -> Optional[bool]:
return self._data.get("cacheEnabled")
@property
def legacy_polygons(self) -> Optional[bool]:
return self._data.get("legacyPolygons")
@property
def estimates(self) -> Optional[bool]:
return self._data.get("estimates")
@property
def analyzer(self) -> Optional[str]:
return self._data.get("analyzer")
@property
def cleanup_interval_step(self) -> Optional[int]:
return self._data.get("cleanupIntervalStep")
@property
def commit_interval_msec(self) -> Optional[int]:
return self._data.get("commitIntervalMsec")
@property
def consolidation_interval_msec(self) -> Optional[int]:
return self._data.get("consolidationIntervalMsec")
@property
def consolidation_policy(self) -> Optional[Json]:
return self._data.get("consolidationPolicy")
@property
def primary_sort(self) -> Optional[Json]:
return self._data.get("primarySort")
@property
def stored_values(self) -> Optional[List[Any]]:
return self._data.get("storedValues")
@property
def write_buffer_active(self) -> Optional[int]:
return self._data.get("writeBufferActive")
@property
def write_buffer_idle(self) -> Optional[int]:
return self._data.get("writeBufferIdle")
@property
def write_buffer_size_max(self) -> Optional[int]:
return self._data.get("writeBufferSizeMax")
@property
def primary_key_cache(self) -> Optional[bool]:
return self._data.get("primaryKeyCache")
@property
def parallelism(self) -> Optional[int]:
return self._data.get("parallelism")
@property
def optimize_top_k(self) -> Optional[List[str]]:
return self._data.get("optimizeTopK")
@property
def track_list_positions(self) -> Optional[bool]:
return self._data.get("trackListPositions")
@property
def version(self) -> Optional[int]:
return self._data.get("version")
@property
def include_all_fields(self) -> Optional[bool]:
return self._data.get("includeAllFields")
@property
def features(self) -> Optional[List[str]]:
return self._data.get("features")
@property
def error_message(self) -> Optional[str]:
return self._data.get("errorMessage")
@property
def training_state(self) -> Optional[str]:
return self._data.get("trainingState")
[docs]
class QueryProperties(JsonWrapper):
"""Extra options for AQL queries.
Args:
allow_dirty_reads (bool | None): If set to `True`, when executing the query
against a cluster deployment, the Coordinator is allowed to read from any
shard replica and not only from the leader.
allow_retry (bool | None): Setting it to `True` makes it possible to retry
fetching the latest batch from a cursor.
fail_on_warning (bool | None): If set to `True`, the query will throw an
exception and abort instead of producing a warning.
fill_block_cache (bool | None): If set to `True`, it will make the query
store the data it reads via the RocksDB storage engine in the RocksDB
block cache.
full_count (bool | None): If set to `True` and the query contains a LIMIT
clause, then the result will have some extra attributes.
intermediate_commit_count (int | None): The maximum number of operations
after which an intermediate commit is performed automatically.
intermediate_commit_size (int | None): The maximum total size of operations
after which an intermediate commit is performed automatically.
max_dnf_condition_members (int | None): A threshold for the maximum number of
OR sub-nodes in the internal representation of an AQL FILTER condition.
max_nodes_per_callstack (int | None): The number of execution nodes in the
query plan after that stack splitting is performed to avoid a potential
stack overflow.
max_number_of_plans (int | None): Limits the maximum number of plans that
are created by the AQL query optimizer.
max_runtime (float | None): The query has to be executed within the given
runtime or it is killed. The value is specified in seconds. If unspecified,
there will be no timeout.
max_transaction_size (int | None): The maximum transaction size in bytes.
max_warning_count (int | None): Limits the maximum number of warnings a
query will return.
optimizer (dict | None): Options related to the query optimizer.
profile (int | None): Return additional profiling information in the query
result. Can be set to 1 or 2 (for more detailed profiling information).
satellite_sync_wait (flat | None): How long a DB-Server has time to bring
the SatelliteCollections involved in the query into sync (in seconds).
skip_inaccessible_collections (bool | None): Treat collections to which a user
has no access rights for as if these collections are empty.
spill_over_threshold_memory_usage (int | None): This option allows queries to
store intermediate and final results temporarily on disk if the amount of
memory used (in bytes) exceeds the specified value.
spill_over_threshold_num_rows (int | None): This option allows queries to
store intermediate and final results temporarily on disk if the number
of rows produced by the query exceeds the specified value.
stream (bool | None): Can be enabled to execute the query lazily.
use_plan_cache (bool | None): Set this option to `True` to utilize
a cached query plan or add the execution plan of this query to the
cache if it’s not in the cache yet.
Example:
.. code-block:: json
{
"maxPlans": 1,
"optimizer": {
"rules": [
"-all",
"+remove-unnecessary-filters"
]
}
}
References:
- `create-a-cursor <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#create-a-cursor_body_options>`__
""" # noqa: E501
def __init__(
self,
allow_dirty_reads: Optional[bool] = None,
allow_retry: Optional[bool] = None,
fail_on_warning: Optional[bool] = None,
fill_block_cache: Optional[bool] = None,
full_count: Optional[bool] = None,
intermediate_commit_count: Optional[int] = None,
intermediate_commit_size: Optional[int] = None,
max_dnf_condition_members: Optional[int] = None,
max_nodes_per_callstack: Optional[int] = None,
max_number_of_plans: Optional[int] = None,
max_runtime: Optional[Number] = None,
max_transaction_size: Optional[int] = None,
max_warning_count: Optional[int] = None,
optimizer: Optional[Json] = None,
profile: Optional[int] = None,
satellite_sync_wait: Optional[Number] = None,
skip_inaccessible_collections: Optional[bool] = None,
spill_over_threshold_memory_usage: Optional[int] = None,
spill_over_threshold_num_rows: Optional[int] = None,
stream: Optional[bool] = None,
use_plan_cache: Optional[bool] = None,
) -> None:
data: Json = dict()
if allow_dirty_reads is not None:
data["allowDirtyReads"] = allow_dirty_reads
if allow_retry is not None:
data["allowRetry"] = allow_retry
if fail_on_warning is not None:
data["failOnWarning"] = fail_on_warning
if fill_block_cache is not None:
data["fillBlockCache"] = fill_block_cache
if full_count is not None:
data["fullCount"] = full_count
if intermediate_commit_count is not None:
data["intermediateCommitCount"] = intermediate_commit_count
if intermediate_commit_size is not None:
data["intermediateCommitSize"] = intermediate_commit_size
if max_dnf_condition_members is not None:
data["maxDNFConditionMembers"] = max_dnf_condition_members
if max_nodes_per_callstack is not None:
data["maxNodesPerCallstack"] = max_nodes_per_callstack
if max_number_of_plans is not None:
data["maxNumberOfPlans"] = max_number_of_plans
if max_runtime is not None:
data["maxRuntime"] = max_runtime
if max_transaction_size is not None:
data["maxTransactionSize"] = max_transaction_size
if max_warning_count is not None:
data["maxWarningCount"] = max_warning_count
if optimizer is not None:
data["optimizer"] = optimizer
if profile is not None:
data["profile"] = profile
if satellite_sync_wait is not None:
data["satelliteSyncWait"] = satellite_sync_wait
if skip_inaccessible_collections is not None:
data["skipInaccessibleCollections"] = skip_inaccessible_collections
if spill_over_threshold_memory_usage is not None:
data["spillOverThresholdMemoryUsage"] = spill_over_threshold_memory_usage
if spill_over_threshold_num_rows is not None:
data["spillOverThresholdNumRows"] = spill_over_threshold_num_rows
if stream is not None:
data["stream"] = stream
if use_plan_cache is not None:
data["usePlanCache"] = use_plan_cache
super().__init__(data)
@property
def allow_dirty_reads(self) -> Optional[bool]:
return self._data.get("allowDirtyReads")
@property
def allow_retry(self) -> Optional[bool]:
return self._data.get("allowRetry")
@property
def fail_on_warning(self) -> Optional[bool]:
return self._data.get("failOnWarning")
@property
def fill_block_cache(self) -> Optional[bool]:
return self._data.get("fillBlockCache")
@property
def full_count(self) -> Optional[bool]:
return self._data.get("fullCount")
@property
def intermediate_commit_count(self) -> Optional[int]:
return self._data.get("intermediateCommitCount")
@property
def intermediate_commit_size(self) -> Optional[int]:
return self._data.get("intermediateCommitSize")
@property
def max_dnf_condition_members(self) -> Optional[int]:
return self._data.get("maxDNFConditionMembers")
@property
def max_nodes_per_callstack(self) -> Optional[int]:
return self._data.get("maxNodesPerCallstack")
@property
def max_number_of_plans(self) -> Optional[int]:
return self._data.get("maxNumberOfPlans")
@property
def max_runtime(self) -> Optional[Number]:
return self._data.get("maxRuntime")
@property
def max_transaction_size(self) -> Optional[int]:
return self._data.get("maxTransactionSize")
@property
def max_warning_count(self) -> Optional[int]:
return self._data.get("maxWarningCount")
@property
def optimizer(self) -> Optional[Json]:
return self._data.get("optimizer")
@property
def profile(self) -> Optional[int]:
return self._data.get("profile")
@property
def satellite_sync_wait(self) -> Optional[Number]:
return self._data.get("satelliteSyncWait")
@property
def skip_inaccessible_collections(self) -> Optional[bool]:
return self._data.get("skipInaccessibleCollections")
@property
def spill_over_threshold_memory_usage(self) -> Optional[int]:
return self._data.get("spillOverThresholdMemoryUsage")
@property
def spill_over_threshold_num_rows(self) -> Optional[int]:
return self._data.get("spillOverThresholdNumRows")
@property
def stream(self) -> Optional[bool]:
return self._data.get("stream")
@property
def use_plan_cache(self) -> Optional[bool]:
return self._data.get("usePlanCache")
[docs]
class QueryExecutionPlan(JsonWrapper):
"""The execution plan of an AQL query.
References:
- `plan <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#create-a-cursor_res_201_extra_plan>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def collections(self) -> Optional[Jsons]:
return self._data.get("collections")
@property
def estimated_cost(self) -> Optional[float]:
return self._data.get("estimatedCost")
@property
def estimated_nr_items(self) -> Optional[int]:
return self._data.get("estimatedNrItems")
@property
def is_modification_query(self) -> Optional[bool]:
return self._data.get("isModificationQuery")
@property
def nodes(self) -> Optional[Jsons]:
return self._data.get("nodes")
@property
def rules(self) -> Optional[List[str]]:
return self._data.get("rules")
@property
def variables(self) -> Optional[Jsons]:
return self._data.get("variables")
[docs]
class QueryExecutionProfile(JsonWrapper):
"""The duration of the different query execution phases in seconds.
Example:
.. code-block:: json
{
"initializing" : 0.0000028529999838156073,
"parsing" : 0.000029285000010759177,
"optimizing ast" : 0.0000040699999885873694,
"loading collections" : 0.000012807000018710823,
"instantiating plan" : 0.00002348999998957879,
"optimizing plan" : 0.00006598600000984334,
"instantiating executors" : 0.000027471999999306718,
"executing" : 0.7550992429999894,
"finalizing" : 0.00004103500000951499
}
References:
- `profile <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#create-a-cursor_res_201_extra_profile>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def executing(self) -> Optional[float]:
return self._data.get("executing")
@property
def finalizing(self) -> Optional[float]:
return self._data.get("finalizing")
@property
def initializing(self) -> Optional[float]:
return self._data.get("initializing")
@property
def instantiating_executors(self) -> Optional[float]:
return self._data.get("instantiating executors")
@property
def instantiating_plan(self) -> Optional[float]:
return self._data.get("instantiating plan")
@property
def loading_collections(self) -> Optional[float]:
return self._data.get("loading collections")
@property
def optimizing_ast(self) -> Optional[float]:
return self._data.get("optimizing ast")
@property
def optimizing_plan(self) -> Optional[float]:
return self._data.get("optimizing plan")
@property
def parsing(self) -> Optional[float]:
return self._data.get("parsing")
[docs]
class QueryExecutionStats(JsonWrapper):
"""Statistics of an AQL query.
Example:
.. code-block:: json
{
"writesExecuted" : 0,
"writesIgnored" : 0,
"documentLookups" : 0,
"seeks" : 0,
"scannedFull" : 2,
"scannedIndex" : 0,
"cursorsCreated" : 0,
"cursorsRearmed" : 0,
"cacheHits" : 0,
"cacheMisses" : 0,
"filtered" : 0,
"httpRequests" : 0,
"executionTime" : 0.00019362399999067748,
"peakMemoryUsage" : 0,
"intermediateCommits" : 0
}
References:
- `stats <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#create-a-cursor_res_201_extra_stats>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def cache_hits(self) -> Optional[int]:
return self._data.get("cacheHits")
@property
def cache_misses(self) -> Optional[int]:
return self._data.get("cacheMisses")
@property
def cursors_created(self) -> Optional[int]:
return self._data.get("cursorsCreated")
@property
def cursors_rearmed(self) -> Optional[int]:
return self._data.get("cursorsRearmed")
@property
def document_lookups(self) -> Optional[int]:
return self._data.get("documentLookups")
@property
def execution_time(self) -> Optional[float]:
return self._data.get("executionTime")
@property
def filtered(self) -> Optional[int]:
return self._data.get("filtered")
@property
def full_count(self) -> Optional[int]:
return self._data.get("fullCount")
@property
def http_requests(self) -> Optional[int]:
return self._data.get("httpRequests")
@property
def intermediate_commits(self) -> Optional[int]:
return self._data.get("intermediateCommits")
@property
def nodes(self) -> Optional[Jsons]:
return self._data.get("nodes")
@property
def peak_memory_usage(self) -> Optional[int]:
return self._data.get("peakMemoryUsage")
@property
def scanned_full(self) -> Optional[int]:
return self._data.get("scannedFull")
@property
def scanned_index(self) -> Optional[int]:
return self._data.get("scannedIndex")
@property
def seeks(self) -> Optional[int]:
return self._data.get("seeks")
@property
def writes_executed(self) -> Optional[int]:
return self._data.get("writesExecuted")
@property
def writes_ignored(self) -> Optional[int]:
return self._data.get("writesIgnored")
[docs]
class QueryTrackingConfiguration(JsonWrapper):
"""AQL query tracking configuration.
Example:
.. code-block:: json
{
"enabled": true,
"trackSlowQueries": true,
"trackBindVars": true,
"maxSlowQueries": 64,
"slowQueryThreshold": 10,
"slowStreamingQueryThreshold": 10,
"maxQueryStringLength": 4096
}
References:
- `get-the-aql-query-tracking-configuration <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#get-the-aql-query-tracking-configuration>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def enabled(self) -> bool:
return cast(bool, self._data["enabled"])
@property
def track_slow_queries(self) -> bool:
return cast(bool, self._data["trackSlowQueries"])
@property
def track_bind_vars(self) -> bool:
return cast(bool, self._data["trackBindVars"])
@property
def max_slow_queries(self) -> int:
return cast(int, self._data["maxSlowQueries"])
@property
def slow_query_threshold(self) -> int:
return cast(int, self._data["slowQueryThreshold"])
@property
def slow_streaming_query_threshold(self) -> Optional[int]:
return self._data.get("slowStreamingQueryThreshold")
@property
def max_query_string_length(self) -> int:
return cast(int, self._data["maxQueryStringLength"])
[docs]
class QueryExplainOptions(JsonWrapper):
"""Options for explaining an AQL query.
Args:
all_plans (bool | None): If set to `True`, all possible execution plans are
returned.
max_plans (int | None): The maximum number of plans to return.
optimizer (dict | None): Options related to the query optimizer.
Example:
.. code-block:: json
{
"allPlans" : false,
"maxNumberOfPlans" : 1,
"optimizer" : {
"rules" : [
"-all",
"+use-indexe-for-sort"
]
}
}
References:
- `explain-an-aql-query <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-queries/#explain-an-aql-query>`__
""" # noqa: E501
def __init__(
self,
all_plans: Optional[bool] = None,
max_plans: Optional[int] = None,
optimizer: Optional[Json] = None,
) -> None:
data: Json = dict()
if all_plans is not None:
data["allPlans"] = all_plans
if max_plans is not None:
data["maxNumberOfPlans"] = max_plans
if optimizer is not None:
data["optimizer"] = optimizer
super().__init__(data)
@property
def all_plans(self) -> Optional[bool]:
return self._data.get("allPlans")
@property
def max_plans(self) -> Optional[int]:
return self._data.get("maxNumberOfPlans")
@property
def optimizer(self) -> Optional[Json]:
return self._data.get("optimizer")
[docs]
class QueryCacheProperties(JsonWrapper):
"""AQL Cache Configuration.
Example:
.. code-block:: json
{
"mode" : "demand",
"maxResults" : 128,
"maxResultsSize" : 268435456,
"maxEntrySize" : 16777216,
"includeSystem" : false
}
References:
- `get-the-aql-query-results-cache-configuration <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-query-results-cache/#get-the-aql-query-results-cache-configuration>`__
- `set-the-aql-query-results-cache-configuration <https://docs.arango.ai/arangodb/stable/develop/http-api/queries/aql-query-results-cache/#set-the-aql-query-results-cache-configuration>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def mode(self) -> str:
return cast(str, self._data.get("mode", ""))
@property
def max_results(self) -> int:
return cast(int, self._data.get("maxResults", 0))
@property
def max_results_size(self) -> int:
return cast(int, self._data.get("maxResultsSize", 0))
@property
def max_entry_size(self) -> int:
return cast(int, self._data.get("maxEntrySize", 0))
@property
def include_system(self) -> bool:
return cast(bool, self._data.get("includeSystem", False))
[docs]
class GraphProperties(JsonWrapper):
"""Graph properties.
Example:
.. code-block:: json
{
"_key" : "myGraph",
"edgeDefinitions" : [
{
"collection" : "edges",
"from" : [
"startVertices"
],
"to" : [
"endVertices"
]
}
],
"orphanCollections" : [ ],
"_rev" : "_jJdpHEy--_",
"_id" : "_graphs/myGraph",
"name" : "myGraph"
}
References:
- `get-a-graph <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#get-a-graph>`__
- `list-all-graphs <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#list-all-graphs>`__
- `create-a-graph <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#create-a-graph>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def name(self) -> str:
return cast(str, self._data["name"])
@property
def is_smart(self) -> bool:
"""Check if the graph is a smart graph."""
return cast(bool, self._data.get("isSmart", False))
@property
def is_satellite(self) -> bool:
"""Check if the graph is a satellite graph."""
return cast(bool, self._data.get("isSatellite", False))
@property
def number_of_shards(self) -> Optional[int]:
return cast(Optional[int], self._data.get("numberOfShards"))
@property
def replication_factor(self) -> Optional[int | str]:
return cast(Optional[int | str], self._data.get("replicationFactor"))
@property
def min_replication_factor(self) -> Optional[int]:
return cast(Optional[int], self._data.get("minReplicationFactor"))
@property
def write_concern(self) -> Optional[int]:
return cast(Optional[int], self._data.get("writeConcern"))
@property
def edge_definitions(self) -> Jsons:
return cast(Jsons, self._data.get("edgeDefinitions", list()))
@property
def orphan_collections(self) -> List[str]:
return cast(List[str], self._data.get("orphanCollections", list()))
@staticmethod
def compatibility_formatter(data: Json) -> Json:
result: Json = {}
if "_id" in data:
result["id"] = data["_id"]
if "_key" in data:
result["key"] = data["_key"]
if "name" in data:
result["name"] = data["name"]
if "_rev" in data:
result["revision"] = data["_rev"]
if "orphanCollections" in data:
result["orphan_collection"] = data["orphanCollections"]
if "edgeDefinitions" in data:
result["edge_definitions"] = [
{
"edge_collection": edge_definition["collection"],
"from_vertex_collections": edge_definition["from"],
"to_vertex_collections": edge_definition["to"],
}
for edge_definition in data["edgeDefinitions"]
]
if "isSmart" in data:
result["smart"] = data["isSmart"]
if "isDisjoint" in data:
result["disjoint"] = data["isDisjoint"]
if "isSatellite" in data:
result["is_satellite"] = data["isSatellite"]
if "smartGraphAttribute" in data:
result["smart_field"] = data["smartGraphAttribute"]
if "numberOfShards" in data:
result["shard_count"] = data["numberOfShards"]
if "replicationFactor" in data:
result["replication_factor"] = data["replicationFactor"]
if "minReplicationFactor" in data:
result["min_replication_factor"] = data["minReplicationFactor"]
if "writeConcern" in data:
result["write_concern"] = data["writeConcern"]
return result
[docs]
class GraphOptions(JsonWrapper):
"""Special options for graph creation.
Args:
number_of_shards (int): The number of shards that is used for every
collection within this graph. Cannot be modified later.
replication_factor (int | str): The replication factor used when initially
creating collections for this graph. Can be set to "satellite" to create
a SatelliteGraph, which then ignores `numberOfShards`,
`minReplicationFactor`, and `writeConcern` (Enterprise Edition only).
satellites (list[str] | None): An array of collection names that is used to
create SatelliteCollections for a (Disjoint) SmartGraph using
SatelliteCollections (Enterprise Edition only). Each array element must
be a string and a valid collection name.
smart_graph_attribute (str | None): The attribute name that is used to
smartly shard the vertices of a graph. Only available in
Enterprise Edition.
write_concern (int | None): The write concern for new collections in the
graph.
References:
- `create-a-graph <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#create-a-graph>`__
""" # noqa: E501
def __init__(
self,
number_of_shards: Optional[int] = None,
replication_factor: Optional[int | str] = None,
satellites: Optional[List[str]] = None,
smart_graph_attribute: Optional[str] = None,
write_concern: Optional[int] = None,
) -> None:
data: Json = dict()
if number_of_shards is not None:
data["numberOfShards"] = number_of_shards
if replication_factor is not None:
data["replicationFactor"] = replication_factor
if satellites is not None:
data["satellites"] = satellites
if smart_graph_attribute is not None:
data["smartGraphAttribute"] = smart_graph_attribute
if write_concern is not None:
data["writeConcern"] = write_concern
super().__init__(data)
@property
def number_of_shards(self) -> Optional[int]:
return cast(int, self._data.get("numberOfShards"))
@property
def replication_factor(self) -> Optional[int | str]:
return cast(int | str, self._data.get("replicationFactor"))
@property
def satellites(self) -> Optional[List[str]]:
return cast(Optional[List[str]], self._data.get("satellites"))
@property
def smart_graph_attribute(self) -> Optional[str]:
return cast(Optional[str], self._data.get("smartGraphAttribute"))
@property
def write_concern(self) -> Optional[int]:
return cast(Optional[int], self._data.get("writeConcern"))
[docs]
class VertexCollectionOptions(JsonWrapper):
"""Special options for vertex collection creation.
Args:
satellites (list): An array of collection names that is used to create
SatelliteCollections for a (Disjoint) SmartGraph using
SatelliteCollections (Enterprise Edition only). Each array element must
be a string and a valid collection name.
References:
- `add-a-vertex-collection <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#add-a-vertex-collection>`__
""" # noqa: E501
def __init__(
self,
satellites: Optional[List[str]] = None,
) -> None:
data: Json = dict()
if satellites is not None:
data["satellites"] = satellites
super().__init__(data)
@property
def satellites(self) -> Optional[List[str]]:
return cast(Optional[List[str]], self._data.get("satellites"))
[docs]
class EdgeDefinitionOptions(JsonWrapper):
"""Special options for edge definition creation.
Args:
satellites (list): An array of collection names that is used to create
SatelliteCollections for a (Disjoint) SmartGraph using
SatelliteCollections (Enterprise Edition only). Each array element must
be a string and a valid collection name.
References:
- `add-an-edge-definition <https://docs.arango.ai/arangodb/stable/develop/http-api/graphs/named-graphs/#add-an-edge-definition>`__
""" # noqa: E501
def __init__(
self,
satellites: Optional[List[str]] = None,
) -> None:
data: Json = dict()
if satellites is not None:
data["satellites"] = satellites
super().__init__(data)
@property
def satellites(self) -> Optional[List[str]]:
return cast(Optional[List[str]], self._data.get("satellites"))
[docs]
class AccessToken(JsonWrapper):
"""User access token.
Example:
.. code-block:: json
{
"id" : 1,
"name" : "Token for Service A",
"valid_until" : 1782864000,
"created_at" : 1765543306,
"fingerprint" : "v1...71227d",
"active" : true,
"token" : "v1.7b2265223a3137471227d"
}
References:
- `create-an-access-token <https://docs.arango.ai/arangodb/arangodb/stable/develop/http-api/authentication/#create-an-access-token>`__
""" # noqa: E501
def __init__(self, data: Json) -> None:
super().__init__(data)
@property
def active(self) -> bool:
return cast(bool, self._data["active"])
@property
def created_at(self) -> int:
return cast(int, self._data["created_at"])
@property
def fingerprint(self) -> str:
return cast(str, self._data["fingerprint"])
@property
def id(self) -> int:
return cast(int, self._data["id"])
@property
def name(self) -> str:
return cast(str, self._data["name"])
@property
def token(self) -> str:
return cast(str, self._data["token"])
@property
def valid_until(self) -> int:
return cast(int, self._data["valid_until"])