Read data items from the data reference handler and return as a list.

    Data items will have persistent_object_context set upon return, but caller will need to call finish_reading
    on each of the data items.
512148
def read_library(persistent_storage_system, ignore_older_files) -> typing.Dict:
    
    data_item_uuids = set()
    utilized_deletions = set()  # the uuid's skipped due to being deleted
    deletions = list()

    reader_info_list, library_updates = auto_migrate_storage_system(persistent_storage_system=persistent_storage_system,
                                                                    new_persistent_storage_system=persistent_storage_system,
                                                                    data_item_uuids=data_item_uuids,
                                                                    deletions=deletions,
                                                                    utilized_deletions=utilized_deletions,
                                                                    ignore_older_files=ignore_older_files)

    # next, for each auto migration, create a temporary storage system and read items from that storage system
    # using auto_migrate_storage_system. the data items returned will have been copied to the current storage
    # system (persistent object context).
    for auto_migration in reversed(persistent_storage_system.get_auto_migrations()):
        old_persistent_storage_system = FileStorageSystem(auto_migration.library_path, auto_migration.paths) if auto_migration.paths else auto_migration.storage_system
        new_reader_info_list, new_library_updates = auto_migrate_storage_system(persistent_storage_system=old_persistent_storage_system,
                                                                                new_persistent_storage_system=persistent_storage_system,
                                                                                data_item_uuids=data_item_uuids,
                                                                                deletions=deletions,
                                                                                utilized_deletions=utilized_deletions,
                                                                                ignore_older_files=ignore_older_files)
        reader_info_list.extend(new_reader_info_list)
        library_updates.update(new_library_updates)

    assert len(reader_info_list) == len(data_item_uuids)

    library_storage_properties = persistent_storage_system.library_storage_properties

    for reader_info in reader_info_list:
        properties = reader_info.properties
        properties = Utility.clean_dict(copy.deepcopy(properties) if properties else dict())
        version = properties.get("version", 0)
        if version == DataItem.DataItem.writer_version:
            data_item_uuid = uuid.UUID(properties.get("uuid", uuid.uuid4()))
            library_update = library_updates.get(data_item_uuid, dict())
            library_storage_properties.setdefault("connections", list()).extend(library_update.get("connections", list()))
            library_storage_properties.setdefault("computations", list()).extend(library_update.get("computations", list()))
            library_storage_properties.setdefault("display_items", list()).extend(library_update.get("display_items", list()))

    # mark deletions that need to be tracked because they've been deleted but are also present in older libraries
    # and would be migrated during reading unless they explicitly are prevented from doing so (via data_item_deletions).
    # utilized deletions are the ones that were attempted; if nothing was attempted, then no reason to track it anymore
    # since there is nothing to migrate in the future.
    library_storage_properties["data_item_deletions"] = [str(uuid_) for uuid_ in utilized_deletions]

    connections_list = library_storage_properties.get("connections", list())
    assert len(connections_list) == len({connection.get("uuid") for connection in connections_list})

    computations_list = library_storage_properties.get("computations", list())
    assert len(computations_list) == len({computation.get("uuid") for computation in computations_list})

    # migrations

    if library_storage_properties.get("version", 0) < 2:
        for data_group_properties in library_storage_properties.get("data_groups", list()):
            data_group_properties.pop("data_groups")
            display_item_references = data_group_properties.setdefault("display_item_references", list())
            data_item_uuid_strs = data_group_properties.pop("data_item_uuids", list())
            for data_item_uuid_str in data_item_uuid_strs:
                for display_item_properties in library_storage_properties.get("display_items", list()):
                    data_item_references = [d.get("data_item_reference", None) for d in display_item_properties.get("display_data_channels", list())]
                    if data_item_uuid_str in data_item_references:
                        display_item_references.append(display_item_properties["uuid"])
        data_item_uuid_to_display_item_uuid_map = dict()
        data_item_uuid_to_display_item_dict_map = dict()
        display_to_display_item_map = dict()
        display_to_display_data_channel_map = dict()
        for display_item_properties in library_storage_properties.get("display_items", list()):
            display_to_display_item_map[display_item_properties["display"]["uuid"]] = display_item_properties["uuid"]
            display_to_display_data_channel_map[display_item_properties["display"]["uuid"]] = display_item_properties["display_data_channels"][0]["uuid"]
            data_item_references = [d.get("data_item_reference", None) for d in display_item_properties.get("display_data_channels", list())]
            for data_item_uuid_str in data_item_references:
                data_item_uuid_to_display_item_uuid_map.setdefault(data_item_uuid_str, display_item_properties["uuid"])
                data_item_uuid_to_display_item_dict_map.setdefault(data_item_uuid_str, display_item_properties)
            display_item_properties.pop("display", None)
        for workspace_properties in library_storage_properties.get("workspaces", list()):
            def replace1(d):
                if "children" in d:
                    for dd in d["children"]:
                        replace1(dd)
                if "data_item_uuid" in d:
                    data_item_uuid_str = d.pop("data_item_uuid")
                    display_item_uuid_str = data_item_uuid_to_display_item_uuid_map.get(data_item_uuid_str)
                    if display_item_uuid_str:
                        d["display_item_uuid"] = display_item_uuid_str
            replace1(workspace_properties["layout"])
        for connection_dict in library_storage_properties.get("connections", list()):
            source_uuid_str = connection_dict["source_uuid"]
            if connection_dict["type"] == "interval-list-connection":
                connection_dict["source_uuid"] = display_to_display_item_map.get(source_uuid_str, None)
            if connection_dict["type"] == "property-connection" and connection_dict["source_property"] == "slice_interval":
                connection_dict["source_uuid"] = display_to_display_data_channel_map.get(source_uuid_str, None)

        def fix_specifier(specifier_dict):
            if specifier_dict.get("type") in ("data_item", "display_xdata", "cropped_xdata", "cropped_display_xdata", "filter_xdata", "filtered_xdata"):
                if specifier_dict.get("uuid") in data_item_uuid_to_display_item_dict_map:
                    specifier_dict["uuid"] = data_item_uuid_to_display_item_dict_map[specifier_dict["uuid"]]["display_data_channels"][0]["uuid"]
                else:
                    specifier_dict.pop("uuid", None)
            if specifier_dict.get("type") == "data_item":
                specifier_dict["type"] = "data_source"
            if specifier_dict.get("type") == "data_item_object":
                specifier_dict["type"] = "data_item"
            if specifier_dict.get("type") == "region":
                specifier_dict["type"] = "graphic"

        for computation_dict in library_storage_properties.get("computations", list()):
            for variable_dict in computation_dict.get("variables", list()):
                if "specifier" in variable_dict:
                    specifier_dict = variable_dict["specifier"]
                    if specifier_dict is not None:
                        fix_specifier(specifier_dict)
                if "secondary_specifier" in variable_dict:
                    specifier_dict = variable_dict["secondary_specifier"]
                    if specifier_dict is not None:
                        fix_specifier(specifier_dict)
            for result_dict in computation_dict.get("results", list()):
                fix_specifier(result_dict["specifier"])

        library_storage_properties["version"] = DocumentModel.DocumentModel.library_version

    # TODO: add consistency checks: no duplicated items [by uuid] such as connections or computations or data items

    assert library_storage_properties["version"] == DocumentModel.DocumentModel.library_version

    persistent_storage_system.rewrite_properties(library_storage_properties)

    properties = copy.deepcopy(library_storage_properties)

    for reader_info in reader_info_list:
        data_item_properties = Utility.clean_dict(reader_info.properties if reader_info.properties else dict())
        if data_item_properties.get("version", 0) == DataItem.DataItem.writer_version:
            data_item_properties["__large_format"] = reader_info.large_format
            data_item_properties["__identifier"] = reader_info.identifier
            properties.setdefault("data_items", list()).append(data_item_properties)

    def data_item_created(data_item_properties: typing.Mapping) -> str:
        return data_item_properties.get("created", "1900-01-01T00:00:00.000000")

    properties["data_items"] = sorted(properties.get("data_items", list()), key=data_item_created)

    return properties
1027423
public List<DataSlice> getPersistentData() throws PersistentDataEncodingException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getPersistentData");

    // this is going to be spilled, so if not already told ConsumerKey that it is locked
    // tell it now!
    synchronized (this)
    {
      if (!informedConsumerKeyThatLocked)
      {
        RemoteDispatchableKey dkey = key.getRemoteDispatchableKey();
        RemoteQPConsumerKey ck = null;
        if (dkey instanceof RemoteQPConsumerKey)
          ck = (RemoteQPConsumerKey) dkey;
        if (ck != null)
        {
          ck.messageLocked(key);
          informedConsumerKeyThatLocked = true;
        }
      }
      // set RemoteDispatchableKey to null so no one can inform the RemoteDispatchableKey about any state change
      // in this message in the future. If we did not do this, it is possible that an eventUnlocked()
      // call occurs after this method call and before the in-memory AIMessageItem object is discarded,
      // which would cause informedConsumerKeyThatLocked to be set to false and the ConsumerKey unlocked
      // count would be incremented.
      key.clearRemoteDispatchableKey();
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getPersistentData");

    return super.getPersistentData();
  }
912866
def load_fetched_objects_into_contexts(events, model_data, context_hints_per_source):
    
    for event in events:
        context_hints = context_hints_per_source.get(event.source, {})
        for context_key, hints in context_hints.items():
            model = get_model(hints['app_name'], hints['model_name'])
            for d, value in dict_find(event.context, context_key):
                if isinstance(value, list):
                    for i, model_id in enumerate(d[context_key]):
                        d[context_key][i] = model_data[model].get(model_id)
                else:
                    d[context_key] = model_data[model].get(value)
1143171
private CompletableFuture<Iterator<Operation>> readData(Void ignored) {
        long traceId = LoggerHelpers.traceEnterWithContext(log, this.traceObjectId, "readData");
        try {
            Duration readTimeout = getReadTimeout();
            return this.dataSource
                    .read(this.state.getLastReadSequenceNumber(), this.config.getMaxItemsToReadAtOnce(), readTimeout)
                    .thenApply(result -> {
                        LoggerHelpers.traceLeave(log, this.traceObjectId, "readData", traceId);
                        return result;
                    })
                    .exceptionally(ex -> {
                        ex = Exceptions.unwrap(ex);
                        if (ex instanceof TimeoutException) {
                            // TimeoutExceptions are acceptable for Reads. In that case we just return null as opposed from
                            // killing the entire Iteration. Even if we were unable to read, we may still need to flush
                            // in this iteration or do other tasks.
                            log.debug("{}: Iteration[{}] No items were read during allotted timeout of {}ms", this.traceObjectId, this.state.getIterationId(), readTimeout.toMillis());
                            return null;
                        } else {
                            throw new CompletionException(ex);
                        }
                    });
        } catch (Throwable ex) {
            // This is for synchronous exceptions.
            Throwable realEx = Exceptions.unwrap(ex);
            if (realEx instanceof TimeoutException) {
                logErrorHandled(realEx);
                return CompletableFuture.completedFuture(null);
            } else {
                return Futures.failedFuture(ex);
            }
        }
    }
948616
def _set_backreferences(self, context, items, **kwargs):
        

        # Don't set any references during initialization.
        # This might cause a recursion error when calling `getRaw` to fetch the
        # current set UIDs!
        initializing = kwargs.get('_initializing_', False)
        if initializing:
            return

        # UID of the current object
        uid = api.get_uid(context)
        # current set UIDs
        raw = self.getRaw(context) or []
        # handle single reference fields
        if isinstance(raw, basestring):
            raw = [raw, ]
        cur = set(raw)
        # UIDs to be set
        new = set(map(api.get_uid, items))
        # removed UIDs
        removed = cur.difference(new)

        # Unlink removed UIDs from the source
        for uid in removed:
            source = api.get_object_by_uid(uid, None)
            if source is None:
                logger.warn("UID {} does not exist anymore".format(uid))
                continue
            self.unlink_reference(source, context)

        # Link backrefs
        for item in items:
            self.link_reference(item, context)
331714
private Object processItem(Object itemRead) {
		logger.entering(sourceClass, "processItem", itemRead);
		Object processedItem = null;

		// if no processor defined for this chunk
		if (processorProxy == null){
			return itemRead;
		}

		try {

			// call process listeners before and after the actual process call
			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.beforeProcess(itemRead);
			}

			processedItem = processorProxy.processItem(itemRead);

			if (processedItem == null) {
				currentItemStatus.setFiltered(true);
			}

			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.afterProcess(itemRead, processedItem);
			}
		} catch (Exception e) {
			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.onProcessError(itemRead, e);
			}
			if(!currentChunkStatus.isRetryingAfterRollback()) {
				if (retryProcessException(e, itemRead)) {
					if (!retryHandler.isRollbackException(e)) {
						processedItem = processItem(itemRead);
					} else {
						currentChunkStatus.markForRollbackWithRetry(e);
					}
				}
				else if (skipProcessException(e, itemRead)) {
					currentItemStatus.setSkipped(true);
					stepContext.getMetric(MetricImpl.MetricType.PROCESS_SKIP_COUNT).incValue();
				}
				else {
					throw new BatchContainerRuntimeException(e);
				}
			}
			else {
				if (skipProcessException(e, itemRead)) {
					currentItemStatus.setSkipped(true);
					stepContext.getMetric(MetricImpl.MetricType.PROCESS_SKIP_COUNT).incValue();
				} else if (retryProcessException(e, itemRead)) {

					if (!retryHandler.isRollbackException(e)) {
						// retry without rollback
						processedItem = processItem(itemRead);
					} else {
						// retry with rollback
						currentChunkStatus.markForRollbackWithRetry(e);
					}
				} else {
					throw new BatchContainerRuntimeException(e);
				}
			}

		} catch (Throwable e) {
			throw new BatchContainerRuntimeException(e);
		}

		logger.exiting(sourceClass, "processItem", processedItem==null ? "<null>" : processedItem);
		return processedItem;
	}
182634
def read_all(self):
        
        result = self._data[self._position:]
        self._position = None  # ensure no subsequent read()
        return result
98254
def _read(self):
        """"""
        _, data_img_name, label_img_name = self.f.readline().strip('\n').split("\t")
        data = {}
        label = {}
        data[self.data_name], label[self.label_name] = self._read_img(data_img_name, label_img_name)
        return list(data.items()), list(label.items())
1199373
def import_data(self, data_items):
        
        # keep counts of all actions
        record = {
            'insert': 0, 'update': 0, 'noop': 0,
            'start': utcnow(),
            'records': {
                'insert': [],
                'update': [],
                'noop': [],
            }
        }

        for json_id, data in self._prepare_imports(data_items):
            obj_id, what = self.import_item(data)
            self.json_to_db_id[json_id] = obj_id
            record['records'][what].append(obj_id)
            record[what] += 1

        # all objects are loaded, a perfect time to do inter-object resolution and other tasks
        self.postimport()

        record['end'] = utcnow()

        return {self._type: record}
738890
def data_context(fn, mode="r"):
    
    with open(data_context_name(fn), mode) as f:
        return f.read()
949255
def from_raw_data(cls, item_raw):
        
        deserialized = {}
        for name, value in item_raw.items():
            attr = getattr(cls, name, None)
            if attr is None:
                continue
            deserialized[name] = attr.decode(value)
        return cls(**deserialized)
1344395
def persistent_object_context_changed(self):
        
        super().persistent_object_context_changed()

        def source_registered(source):
            self.__source = source

        def source_unregistered(source=None):
            pass

        def reference_registered(property_name, reference):
            self.__referenced_objects[property_name] = reference

        def reference_unregistered(property_name, reference=None):
            pass

        if self.persistent_object_context:
            self.persistent_object_context.subscribe(self.source_uuid, source_registered, source_unregistered)

            for property_name, value in self.__properties.items():
                if isinstance(value, dict) and value.get("type") in {"data_item", "display_item", "data_source", "graphic", "structure"} and "uuid" in value:
                    self.persistent_object_context.subscribe(uuid.UUID(value["uuid"]), functools.partial(reference_registered, property_name), functools.partial(reference_unregistered, property_name))
        else:
            source_unregistered()

            for property_name, value in self.__properties.items():
                if isinstance(value, dict) and value.get("type") in {"data_item", "display_item", "data_source", "graphic", "structure"} and "uuid" in value:
                    reference_unregistered(property_name)
532405
def read(self):
        
        __result = []

        __ll = self.readline()
        while __ll:
            __result.append(__ll)
            __ll = self.readline()

        return list(__result)
1244924
def data_item_live(self, data_item):
        
        class LiveContextManager:
            def __init__(self, manager, object):
                self.__manager = manager
                self.__object = object
            def __enter__(self):
                self.__manager.begin_data_item_live(self.__object)
                return self
            def __exit__(self, type, value, traceback):
                self.__manager.end_data_item_live(self.__object)
        return LiveContextManager(self, data_item)
298845
def items(self):
        """"""
        if self.scraper.current_item is not self:
            self._move_here()

        if self._items is None:
            self._items = ItemList()
            self._items.scraper = self.scraper
            self._items.collection = self
            for i in self.scraper._fetch_itemslist(self):
                i.parent = self
                if i.type == TYPE_DATASET and i.dialect is None:
                    i.dialect = self.scraper.dialect
                self._items.append(i)
        return self._items
749795
def get_context_data(self, **kwargs):
        """"""
        
        context = super(AnimalListAlive, self).get_context_data(**kwargs)
        context['list_type'] = 'Alive'
        return context
1369321
def _as_in_context(data, ctx):
    """"""
    if isinstance(data, nd.NDArray):
        return data.as_in_context(ctx)
    elif isinstance(data, (list, tuple)):
        return [_as_in_context(d, ctx) for d in data]
    return data