Source code for tiled.client.base

import importlib
import time
from dataclasses import asdict

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

class MetadataRevisions:
    def __init__(self, context, link):
        self._cached_len = None
        self.context = context
        self._link = link

    def __len__(self):

        LENGTH_CACHE_TTL = 1  # second

        now = time.monotonic()
        if self._cached_len is not None:
            length, deadline = self._cached_len
            if now < deadline:
                # Used the cached value and do not make any request.
                return length

        content = self.context.get_json(
            self._link, params={"page[offset]": 0, "page[limit]": 0}
        length = content["meta"]["count"]
        self._cached_len = (length, now + LENGTH_CACHE_TTL)
        return length

    def __getitem__(self, item_):
        self._cached_len = None

        if isinstance(item_, int):
            offset = item_
            limit = 1

            content = self.context.get_json(
                self._link, params={"page[offset]": offset, "page[limit]": limit}

            (result,) = content["data"]
            return result

        elif isinstance(item_, slice):
            offset = item_.start
            if offset is None:
                offset = 0
            if item_.stop is None:
                params = f"?page[offset]={offset}"
                limit = item_.stop - offset
                params = f"?page[offset]={offset}&page[limit]={limit}"

            next_page = self._link + params
            result = []
            while next_page is not None:
                content = self.context.get_json(next_page)
                if len(result) == 0:
                    result = content.copy()
                next_page = content["links"]["next"]

            return result["data"]

    def delete_revision(self, n):
        self.context.delete_content(self._link, None, params={"n": n})

[docs]class BaseClient:
[docs] def __init__(self, context, *, item, structure_clients): self._context = context self._item = item self._cached_len = None # a cache just for __len__ self.structure_clients = structure_clients self._metadata_revisions = None super().__init__()
[docs] def login(self, username=None, 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: >>> c.login() You have ... minutes visit this URL https://... and enter the code: XXXX-XXXX """ self.context.login(username=username, provider=provider)
[docs] def logout(self): """ Log out. This method is idempotent: if you are already logged out, it will do nothing. """ 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 specs(self): "List of specifications describing the structure of the metadata and/or data." return ListView([Spec(**spec) for spec in self._item["attributes"]["specs"]]) @property def references(self): "References (links) to related context, metadata, or data." # We use .get(...) here for backward-compatibility. # This can be a straight dict lookup once all servers are updated. return ListView(self._item["attributes"].get("references", [])) @property def uri(self): "Direct link to this entry" return self.item["links"]["self"] @property def offline(self): return self.context.offline @offline.setter def offline(self, value): self.context.offline = bool(value)
[docs] def new_variation(self, structure_clients=UNCHANGED, **kwargs): """ This is intended primarily for internal use and use by subclasses. """ if structure_clients is UNCHANGED: structure_clients = self.structure_clients return type(self)( item=self._item, structure_clients=structure_clients, **kwargs, )
@property def formats(self): "List formats that the server can export this data as." formats = set() for spec in self.item["attributes"]["specs"]: formats.update(self.context.server_info["formats"].get(spec, [])) formats.update( self.context.server_info["formats"][ self.item["attributes"]["structure_family"] ] ) return sorted(formats) def update_metadata(self, metadata=None, specs=None, references=None): """ EXPERIMENTAL: Update metadata. This is subject to change or removal without notice Parameters ---------- metadata : dict, optional User metadata. May be nested. Must contain only basic types (e.g. numbers, strings, lists, dicts) that are JSON-serializable. specs : List[str], optional List of names that are used to label that the data and/or metadata conform to some named standard specification. references : List[Dict[str, URL]], optional References (e.g. links) to related information. This may include links into other Tiled data sets, search results, or external resources unrelated to Tiled. """ self._cached_len = None if specs is None: normalized_specs = None else: normalized_specs = [] for spec in specs: if isinstance(spec, str): spec = Spec(spec) normalized_specs.append(asdict(spec)) data = { "metadata": metadata, "specs": normalized_specs, "references": references, } content = self.context.put_json(self.item["links"]["self"], data) if metadata is not None: if "metadata" in content: # Metadata was accepted and modified by the specs validator on the server side. # It is updated locally using the new version. self._item["attributes"]["metadata"] = content["metadata"] else: # Metadata was accepted as it si by the server. # It is updated locally with the version submitted buy the client. self._item["attributes"]["metadata"] = metadata if specs is not None: self._item["attributes"]["specs"] = normalized_specs if references is not None: self._item["attributes"]["references"] = references @property def metadata_revisions(self): if self._metadata_revisions is None: link = self.item["links"]["self"].replace( "/node/metadata", "/node/revisions", 1 ) self._metadata_revisions = MetadataRevisions(self.context, link) return self._metadata_revisions
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 getattr(getattr(self._structure, "macro", None), "resizable", None): # 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, "sparse": lambda: importlib.import_module( "...structures.sparse", BaseStructureClient.__module__ ).SparseStructure, } )