import importlib
import time
from dataclasses import asdict
from ..structures.core import Spec, StructureFamily
from ..utils import UNCHANGED, DictView, ListView, OneShotCachedMap, safe_json_dump
from .utils import MSGPACK_MIME_TYPE, handle_error
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 = handle_error(
self.context.http_client.get(
self._link,
headers={"Accept": MSGPACK_MIME_TYPE},
params={"page[offset]": 0, "page[limit]": 0},
)
).json()
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 = handle_error(
self.context.http_client.get(
self._link,
headers={"Accept": MSGPACK_MIME_TYPE},
params={"page[offset]": offset, "page[limit]": limit},
)
).json()
(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}"
else:
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 = handle_error(
self.context.http_client.get(
next_page,
headers={"Accept": MSGPACK_MIME_TYPE},
)
).json()
if len(result) == 0:
result = content.copy()
else:
result["data"].append(content["data"])
next_page = content["links"]["next"]
return result["data"]
def delete_revision(self, n):
handle_error(self.context.http_client.delete(self._link, params={"number": n}))
[docs]class BaseClient:
[docs] def __init__(self, context, *, item, structure_clients, structure=None):
self._context = context
self._item = item
self._cached_len = None # a cache just for __len__
self.structure_clients = structure_clients
self._metadata_revisions = None
attributes = self.item["attributes"]
structure_family = attributes["structure_family"]
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
elif structure_family == StructureFamily.container:
self._structure = None
else:
structure_type = STRUCTURE_TYPES[attributes["structure_family"]]
self._structure = structure_type.from_json(attributes["structure"])
super().__init__()
[docs] def structure(self):
"""
Return a dataclass describing the structure of the data.
"""
if getattr(self._structure, "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
[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 uri(self):
"Direct link to this entry"
return self.item["links"]["self"]
[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):
"""
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.
"""
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,
}
content = handle_error(
self.context.http_client.put(
self.item["links"]["self"], content=safe_json_dump(data)
)
).json()
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
@property
def metadata_revisions(self):
if self._metadata_revisions is None:
link = self.item["links"]["self"].replace("/metadata", "/revisions", 1)
self._metadata_revisions = MetadataRevisions(self.context, link)
return self._metadata_revisions
STRUCTURE_TYPES = OneShotCachedMap(
{
StructureFamily.array: lambda: importlib.import_module(
"...structures.array", BaseClient.__module__
).ArrayStructure,
StructureFamily.awkward: lambda: importlib.import_module(
"...structures.awkward", BaseClient.__module__
).AwkwardStructure,
StructureFamily.table: lambda: importlib.import_module(
"...structures.table", BaseClient.__module__
).TableStructure,
StructureFamily.sparse: lambda: importlib.import_module(
"...structures.sparse", BaseClient.__module__
).SparseStructure,
}
)