"""Assets interface
Access to the assets endpoint.
The user is not expected to use this class directly. It is an attribute of the
:class:`Archivist` class.
For example instantiate an Archivist instance and execute the methods of the class:
.. code-block:: python
with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
authtoken = tokenfile.read().strip()
# Initialize connection to Archivist
arch = Archivist(
"https://app.datatrails.ai",
authtoken,
)
asset = arch.assets.create(...)
"""
from copy import deepcopy
from logging import getLogger
from typing import TYPE_CHECKING, Any
# pylint:disable=cyclic-import # but pylint doesn't understand this feature
from . import confirmer
from .asset import Asset
from .constants import (
ASSET_BEHAVIOURS,
ASSETS_LABEL,
ASSETS_SUBPATH,
CONFIRMATION_STATUS,
)
from .dictmerge import _deepmerge
from .errors import ArchivistBadFieldError, ArchivistNotFoundError
from .utils import selector_signature
if TYPE_CHECKING:
from .archivist import Archivist
LOGGER = getLogger(__name__)
[docs]
class _AssetsPublic:
"""AssetsReader
Access to assets entities using CRUD interface. This class is usually
accessed as an attribute of the Archivist or Public class.
Args:
archivist (Archivist): :class:`Archivist` instance
"""
def __init__(self, archivist_instance: "Archivist"):
self._archivist = archivist_instance
self._public = archivist_instance.public
self._subpath = f"{archivist_instance.root}/{ASSETS_SUBPATH}"
def __str__(self) -> str:
return "AssetsPublic()"
[docs]
def _identity(self, identity: str) -> str:
"""Return fully qualified identity
If public then expect a full url as argument
"""
if self._public:
return identity
return f"{self._subpath}/{identity}"
[docs]
def read(self, identity: str) -> Asset:
"""Read asset
Reads asset.
Args:
identity (str): assets identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxx
Returns:
:class:`Asset` instance
"""
return Asset(**self._archivist.get(self._identity(identity)))
[docs]
class _AssetsRestricted(_AssetsPublic):
"""AssetsRestricted
Access to assets entities using CRUD interface. This class is usually
accessed as an attribute of the Archivist or Public class.
Args:
archivist (Archivist): :class:`Archivist` instance
"""
def __init__(self, archivist_instance: "Archivist"):
super().__init__(archivist_instance)
self._label = f"{self._subpath}/{ASSETS_LABEL}"
self.pending_count: int = 0
def __str__(self) -> str:
return f"AssetsRestricted({self._archivist.url})"
def __params(
self, props: "dict[str, Any]|None", attrs: "dict[str, Any]|None"
) -> "dict[str, Any]":
params = deepcopy(props) if props else {}
if attrs:
params["attributes"] = attrs
return _deepmerge(self._archivist.fixtures.get(f"{ASSETS_LABEL}"), params)
[docs]
def create(
self,
*,
props: "dict[str, Any]|None" = None,
attrs: "dict[str, Any]|None" = None,
confirm: bool = False,
) -> Asset:
"""Create asset
Creates asset with defined properties and attributes.
Args:
props (dict): Properties
attrs (dict): attributes of created asset.
confirm (bool): if True wait for asset to be confirmed.
Returns:
:class:`Asset` instance
"""
LOGGER.debug("Create Asset %s", attrs)
# default behaviours are added first - any set in user-specified fixtures or
# in the method args will overide...
newprops = _deepmerge({"behaviours": ASSET_BEHAVIOURS}, props)
data = self.__params(newprops, attrs)
return self.create_from_data(data, confirm=confirm)
[docs]
def create_from_data(
self, data: "dict[str, Any]", *, confirm: bool = False
) -> Asset:
"""Create asset
Creates asset with request body from data stream.
Suitable for reading data from a file using json.load or yaml.load
Args:
data (dict): request body of asset.
confirm (bool): if True wait for asset to be confirmed.
Returns:
:class:`Asset` instance
"""
asset = Asset(**self._archivist.post(self._label, data))
if not confirm:
return asset
return self.wait_for_confirmation(asset["identity"])
[docs]
def create_if_not_exists(
self, data: "dict[str, Any]", *, confirm: bool = False
) -> "tuple[Asset, bool]":
"""
Creates an asset and associated attachments if asset
does not already exist.
Args:
data (dict): request body of asset.
confirm (bool): if True wait for asset to be confirmed.
A YAML representation of the data argument would be:
.. code-block:: yaml
selector:
- attributes:
- arc_display_name
behaviours
- RecordEvidence
attributes:
arc_display_name: DataTrails Front Door
arc_firmware_version: "1.0"
arc_serial_number: das-j1-01
arc_description: Electronic door entry system to DataTrails France
wavestone_asset_id: paris.france.datatrails.das
attachments:
- filename: functests/test_resources/doors/assets/entry-terminal.jpg
content_type: image/jpg
attachment: terminal entry
The 'selector' value is required and will usually specify the 'arc_display_name' as a
secondary key. The keys in 'selector' must exist in the attributes of the asset.
Returns:
tuple of :class:`Asset` instance, Boolean is True if asset already existed
"""
asset = None
existed = False
data = deepcopy(data)
attachments = data.pop("attachments", None)
selector = data.pop("selector") # must exist
props, attrs = selector_signature(selector, data)
try:
asset = self.read_by_signature(props=props, attrs=attrs)
except ArchivistNotFoundError:
LOGGER.info(
"asset with selector %s,%s does not exist - creating", props, attrs
)
else:
LOGGER.info("asset with selector %s,%s already exists", props, attrs)
return asset, True
# any attachments ?
if attachments is not None:
for a in attachments:
# attempt to get attachment to use as a key
attachment_key = a.get("attachment", None)
if attachment_key is None:
# failing that create a key from filename or url
attachment_key = self._archivist.attachments.get_default_key(a)
data["attributes"][attachment_key] = self._archivist.attachments.create(
a
)
asset = self.create_from_data(
data=data,
confirm=confirm,
)
return asset, existed
[docs]
def wait_for_confirmation(self, identity: str) -> Asset:
"""Wait for asset to be confirmed.
Waits for asset to be confirmed.
Args:
identity (str): identity of asset
Returns:
True if asset is confirmed.
"""
confirmer.MAX_TIME = self._archivist.max_time
# pylint: disable=protected-access
return confirmer._wait_for_confirmation(self, identity)
[docs]
def wait_for_confirmed(
self,
*,
props: "dict[str, Any]|None" = None,
attrs: "dict[str, Any]|None" = None,
) -> bool:
"""Wait for assets to be confirmed.
Waits for all assets that match criteria to be confirmed.
Args:
props (dict): e.g. {"tracked": "TRACKED" }
attrs (dict): e.g. {"arc_display_type": "door" }
Returns:
True if all assets are confirmed.
"""
# check that entities exist
newprops = deepcopy(props) if props else {}
newprops.pop(CONFIRMATION_STATUS, None)
LOGGER.debug("Count assets %s", newprops)
count = self.count(props=newprops, attrs=attrs)
if count == 0:
raise ArchivistNotFoundError("No assets exist")
confirmer.MAX_TIME = self._archivist.max_time
# pylint: disable=protected-access
return confirmer._wait_for_confirmed(self, props=newprops, attrs=attrs)
[docs]
def count(
self,
*,
props: "dict[str, Any]|None" = None,
attrs: "dict[str, Any]|None" = None,
) -> int:
"""Count assets.
Counts number of assets that match criteria.
Args:
props (dict): e.g. {"confirmation_status": "CONFIRMED" }
attrs (dict): e.g. {"arc_display_type": "door" }
Returns:
integer count of assets.
"""
return self._archivist.count(self._label, params=self.__params(props, attrs))
[docs]
def list(
self,
*,
page_size: "int|None" = None,
props: "dict[str, Any]|None" = None,
attrs: "dict[str, Any]|None" = None,
):
"""List assets.
Lists assets that match criteria.
Args:
props (dict): optional e.g. {"tracked": "TRACKED" }
attrs (dict): optional e.g. {"arc_display_type": "door" }
page_size (int): optional page size. (Rarely used).
Returns:
iterable that returns :class:`Asset` instances
"""
return (
Asset(**a)
for a in self._archivist.list(
self._label,
ASSETS_LABEL,
page_size=page_size,
params=self.__params(props, attrs),
)
)
[docs]
def read_by_signature(
self,
*,
props: "dict[str, Any]|None" = None,
attrs: "dict[str, Any]|None" = None,
) -> Asset:
"""Read Asset by signature.
Reads asset that meets criteria. Only one asset is expected.
Args:
props (dict): e.g. {"tracked": "TRACKED" }
attrs (dict): e.g. {"arc_display_type": "door" }
Returns:
:class:`Asset` instance
"""
assets_label = f"public{ASSETS_LABEL}" if self._public else ASSETS_LABEL
return Asset(
**self._archivist.get_by_signature(
self._label,
assets_label,
params=self.__params(props, attrs),
)
)
[docs]
def publicurl(self, identity: str) -> str:
"""Read asset public url
Reads assets public url.
Args:
identity (str): assets identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxx
Returns:
:str: publicurl as string
"""
body = self._archivist.get(f"{self._identity(identity)}:publicurl")
publicurl = body.get("publicurl")
if publicurl is None:
raise ArchivistBadFieldError("No publicurl found in response")
return publicurl