Skip to content
Snippets Groups Projects

API helper for `list_element_parents`

Merged API helper for `list_element_parents`
helper-list-element-parents into master
All threads resolved!
Merged Manon Blanco requested to merge helper-list-element-parents into master
All threads resolved!
Compare
and
3 files
+ 630
− 4
Compare changes
  • Side-by-side
  • Inline
Files
3
arkindex_worker/worker/element.py
+ 162
− 0
@@ -497,3 +497,165 @@ class ElementMixin(object):
)
return children
def list_element_parents(
self,
element: Union[Element, CachedElement],
folder: Optional[bool] = None,
name: Optional[str] = None,
recursive: Optional[bool] = None,
transcription_worker_version: Optional[Union[str, bool]] = None,
transcription_worker_run: Optional[Union[str, bool]] = None,
type: Optional[str] = None,
with_classes: Optional[bool] = None,
with_corpus: Optional[bool] = None,
with_metadata: Optional[bool] = None,
with_has_children: Optional[bool] = None,
with_zone: Optional[bool] = None,
worker_version: Optional[Union[str, bool]] = None,
worker_run: Optional[Union[str, bool]] = None,
) -> Union[Iterable[dict], Iterable[CachedElement]]:
"""
List parents of an element.
:param element: Child element to find parents of.
:param folder: Restrict to or exclude elements with folder types.
This parameter is not supported when caching is enabled.
:param name: Restrict to elements whose name contain a substring (case-insensitive).
This parameter is not supported when caching is enabled.
:param recursive: Look for elements recursively (grand-children, etc.)
This parameter is not supported when caching is enabled.
:param transcription_worker_version: Restrict to elements that have a transcription created by a worker version with this UUID.
This parameter is not supported when caching is enabled.
:param transcription_worker_run: Restrict to elements that have a transcription created by a worker run with this UUID.
This parameter is not supported when caching is enabled.
:param type: Restrict to elements with a specific type slug
This parameter is not supported when caching is enabled.
:param with_classes: Include each element's classifications in the response.
This parameter is not supported when caching is enabled.
:param with_corpus: Include each element's corpus in the response.
This parameter is not supported when caching is enabled.
:param with_has_children: Include the ``has_children`` attribute in the response,
indicating if this element has child elements of its own.
This parameter is not supported when caching is enabled.
:param with_metadata: Include each element's metadata in the response.
This parameter is not supported when caching is enabled.
:param with_zone: Include the ``zone`` attribute in the response,
holding the element's image and polygon.
This parameter is not supported when caching is enabled.
:param worker_version: Restrict to elements created by a worker version with this UUID.
:param worker_run: Restrict to elements created by a worker run with this UUID.
:return: An iterable of dicts from the ``ListElementParents`` API endpoint,
or an iterable of [CachedElement][arkindex_worker.cache.CachedElement] when caching is enabled.
"""
assert element and isinstance(
element, (Element, CachedElement)
), "element shouldn't be null and should be an Element or CachedElement"
query_params = {}
if folder is not None:
assert isinstance(folder, bool), "folder should be of type bool"
query_params["folder"] = folder
if name:
assert isinstance(name, str), "name should be of type str"
query_params["name"] = name
if recursive is not None:
assert isinstance(recursive, bool), "recursive should be of type bool"
query_params["recursive"] = recursive
if transcription_worker_version is not None:
assert isinstance(
transcription_worker_version, (str, bool)
), "transcription_worker_version should be of type str or bool"
if isinstance(transcription_worker_version, bool):
assert (
transcription_worker_version is False
), "if of type bool, transcription_worker_version can only be set to False"
query_params["transcription_worker_version"] = transcription_worker_version
if transcription_worker_run is not None:
assert isinstance(
transcription_worker_run, (str, bool)
), "transcription_worker_run should be of type str or bool"
if isinstance(transcription_worker_run, bool):
assert (
transcription_worker_run is False
), "if of type bool, transcription_worker_run can only be set to False"
query_params["transcription_worker_run"] = transcription_worker_run
if type:
assert isinstance(type, str), "type should be of type str"
query_params["type"] = type
if with_classes is not None:
assert isinstance(with_classes, bool), "with_classes should be of type bool"
query_params["with_classes"] = with_classes
if with_corpus is not None:
assert isinstance(with_corpus, bool), "with_corpus should be of type bool"
query_params["with_corpus"] = with_corpus
if with_has_children is not None:
assert isinstance(
with_has_children, bool
), "with_has_children should be of type bool"
query_params["with_has_children"] = with_has_children
if with_metadata is not None:
assert isinstance(
with_metadata, bool
), "with_metadata should be of type bool"
query_params["with_metadata"] = with_metadata
if with_zone is not None:
assert isinstance(with_zone, bool), "with_zone should be of type bool"
query_params["with_zone"] = with_zone
if worker_version is not None:
assert isinstance(
worker_version, (str, bool)
), "worker_version should be of type str or bool"
if isinstance(worker_version, bool):
assert (
worker_version is False
), "if of type bool, worker_version can only be set to False"
query_params["worker_version"] = worker_version
if worker_run is not None:
assert isinstance(
worker_run, (str, bool)
), "worker_run should be of type str or bool"
if isinstance(worker_run, bool):
assert (
worker_run is False
), "if of type bool, worker_run can only be set to False"
query_params["worker_run"] = worker_run
if self.use_cache:
# Checking that we only received query_params handled by the cache
assert set(query_params.keys()) <= {
"type",
"worker_version",
"worker_run",
}, "When using the local cache, you can only filter by 'type' and/or 'worker_version' and/or 'worker_run'"
parent_ids = CachedElement.select(CachedElement.parent_id).where(
CachedElement.id == element.id
)
query = CachedElement.select().where(CachedElement.id.in_(parent_ids))
if type:
query = query.where(CachedElement.type == type)
if worker_version is not None:
# If worker_version=False, filter by manual worker_version e.g. None
worker_version_id = worker_version or None
if worker_version_id:
query = query.where(
CachedElement.worker_version_id == worker_version_id
)
else:
query = query.where(CachedElement.worker_version_id.is_null())
if worker_run is not None:
# If worker_run=False, filter by manual worker_run e.g. None
worker_run_id = worker_run or None
if worker_run_id:
query = query.where(CachedElement.worker_run_id == worker_run_id)
else:
query = query.where(CachedElement.worker_run_id.is_null())
return query
else:
parents = self.api_client.paginate(
"ListElementParents", id=element.id, **query_params
)
return parents