Source code for tiled.client.base

import importlib

from ..utils import UNCHANGED, DictView, ListView, OneShotCachedMap
from .cache import Revalidate, verify_cache

[docs]class BaseClient:
[docs] def __init__(self, context, *, item, path, params, structure_clients): self._context = context if isinstance(path, str): raise ValueError("path is expected to be a list of segments") # Stash *immutable* copies just to be safe. self._path = tuple(path or []) self._item = item self._cached_len = None # a cache just for __len__ self._params = params or {} self.structure_clients = structure_clients super().__init__()
def login(self, provider=None): """ Depending on the server's authentication method, this will prompt for username/password >>> c.login() Username: USERNAME Password: <input is hidden> or prompt you to open a link in a web browser to login with a third party and paste in a access code >>> c.login() Navigate your web browser to this address to obtain access code: ... Access code (quotes optional): <input is hidden> See also c.context.authenticate() and c.context.reauthenticate(). """ self.context.authenticate(provider=provider) # Do NOT return the tokens that are returned by authenticate(). # This avoids displaying valid refresh tokens into places they might persist, # like Jupyter notebooks. def logout(self): """ Log out. """ self.context.logout() def __repr__(self): return f"<{type(self).__name__}>" @property def context(self): return self._context @property def item(self): "JSON payload describing this item. Mostly for internal use." return self._item @property def metadata(self): "Metadata about this data source." # Ensure this is immutable (at the top level) to help the user avoid # getting the wrong impression that editing this would update anything # persistent. return DictView(self._item["attributes"]["metadata"]) @property def path(self): "Sequence of entry names from the root Tree to this entry" return ListView(self._path) @property def uri(self): "Direct link to this entry" return self.item["links"]["self"] @property def username(self): return self.context.username @property def offline(self): return self.context.offline @offline.setter def offline(self, value): self.context.offline = bool(value)
[docs] def new_variation( self, path=UNCHANGED, params=UNCHANGED, structure_clients=UNCHANGED, **kwargs ): """ This is intended primarily for internal use and use by subclasses. """ if path is UNCHANGED: path = self._path if params is UNCHANGED: params = self._params if structure_clients is UNCHANGED: structure_clients = self.structure_clients return type(self)( item=self._item, path=path, params=params, structure_clients=structure_clients, **kwargs, )
class BaseStructureClient(BaseClient): def __init__(self, *args, structure=None, **kwargs): super().__init__(*args, **kwargs) if structure is not None: # Allow the caller to optionally hand us a structure that is already # parsed from a dict into a structure dataclass. self._structure = structure else: attributes = self.item["attributes"] structure_type = STRUCTURE_TYPES[attributes["structure_family"]] self._structure = structure_type.from_json(attributes["structure"])
[docs] def download(self): """ Download all data into the cache. This causes it to be cached if the context is configured with a cache. """ verify_cache(self.context.cache) repr(self)
[docs] def refresh(self, force=False): """ Refresh cached data for this node. Parameters ---------- force: bool If False, (default) refresh only expired cache entries. If True, refresh all cache entries. """ if force: revalidate = Revalidate.FORCE else: revalidate = Revalidate.IF_EXPIRED with self.context.revalidation(revalidate):
[docs] def structure(self): """ Return a dataclass describing the structure of the data. """ if self._structure.macro.resizable: # In the future, conditionally fetch updated information. raise NotImplementedError( "The server has indicated that this has a dynamic, resizable " "structure and this version of the Tiled Python client cannot " "cope with that." ) return self._structure
STRUCTURE_TYPES = OneShotCachedMap( { "array": lambda: importlib.import_module( "...structures.array", BaseStructureClient.__module__ ).ArrayStructure, "dataframe": lambda: importlib.import_module( "...structures.dataframe", BaseStructureClient.__module__ ).DataFrameStructure, "xarray_data_array": lambda: importlib.import_module( "...structures.xarray", BaseStructureClient.__module__ ).DataArrayStructure, "xarray_dataset": lambda: importlib.import_module( "...structures.xarray", BaseStructureClient.__module__ ).DatasetStructure, } )