From f6d0bc8b50596ef0150da51136fa336b66e81a67 Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 08:48:53 -0400 Subject: [PATCH 1/7] Retire databroker.experimental. --- databroker/experimental/__init__.py | 0 databroker/experimental/schemas.py | 112 ---- databroker/experimental/server_ext.py | 810 -------------------------- databroker/tests/test_experimental.py | 558 ------------------ 4 files changed, 1480 deletions(-) delete mode 100644 databroker/experimental/__init__.py delete mode 100644 databroker/experimental/schemas.py delete mode 100644 databroker/experimental/server_ext.py delete mode 100644 databroker/tests/test_experimental.py diff --git a/databroker/experimental/__init__.py b/databroker/experimental/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/databroker/experimental/schemas.py b/databroker/experimental/schemas.py deleted file mode 100644 index 4549fd283..000000000 --- a/databroker/experimental/schemas.py +++ /dev/null @@ -1,112 +0,0 @@ -from typing import Dict, List, Optional, Union -from datetime import datetime - -import pydantic -import pydantic.generics - -from tiled.server.pydantic_array import ArrayStructure -from tiled.server.pydantic_dataframe import DataFrameStructure -from tiled.server.pydantic_sparse import SparseStructure -from tiled.structures.core import StructureFamily - -from tiled.server.schemas import ReferenceDocument, Spec - -# Map structure family to the associated -# structure model. This is used by the validator. -structure_association = { - StructureFamily.array: ArrayStructure, - StructureFamily.dataframe: DataFrameStructure, - StructureFamily.sparse: SparseStructure, - # StructureFamily.node - # ... -} - - -class BaseDocument(pydantic.BaseModel): - key: str - metadata: Dict - specs: List[Spec] - references: List[ReferenceDocument] - updated_at: datetime - - -class Document(BaseDocument): - structure_family: StructureFamily - structure: Union[ArrayStructure, DataFrameStructure, SparseStructure] - mimetype: str - created_at: datetime - data_blob: Optional[bytes] - data_url: Optional[pydantic.AnyUrl] - - @pydantic.root_validator - def validate_structure_matches_structure_family(cls, values): - actual_structure = values.get("structure") - # Given the structure_family, we know what the structure type should be. - expected_structure_type = structure_association[values.get("structure_family")] - if expected_structure_type == StructureFamily.node: - raise Exception( - f"{expected_structure_type} is not currently supported as a writable structure" - ) - elif not isinstance(actual_structure, expected_structure_type): - raise Exception( - "The expected structure type does not match the received structure type" - ) - return values - - @pydantic.validator("mimetype") - def is_mime_type(cls, v): - m_type, _, _ = v.partition("/") - mime_type_list = set( - [ - "application", - "audio", - "font", - "example", - "image", - "message", - "model", - "multipart", - "text", - "video", - ] - ) - - if m_type not in mime_type_list: - raise ValueError(f"{m_type} is not a valid mime type") - return v - - @pydantic.root_validator - def check_data_source(cls, values): - # Making them optional and setting default values might help to meet these conditions - # with the current data types without getting any conflicts - if values.get("data_blob") is not None and values.get("data_url") is not None: - raise ValueError( - "Not Valid: data_blob and data_url contain values. Use just one" - ) - return values - - -class DocumentRevision(BaseDocument): - revision: int - - @classmethod - def from_document(cls, document, revision): - return cls( - key=document.key, - metadata=document.metadata, - specs=document.specs, - references=document.references, - updated_at=document.updated_at, - revision=revision, - ) - - @classmethod - def from_json(cls, json_doc): - return cls( - key=json_doc["key"], - metadata=json_doc["metadata"], - specs=json_doc["specs"], - references=json_doc["references"], - updated_at=json_doc["updated_at"], - revision=json_doc["revision"], - ) diff --git a/databroker/experimental/server_ext.py b/databroker/experimental/server_ext.py deleted file mode 100644 index 27ba27f96..000000000 --- a/databroker/experimental/server_ext.py +++ /dev/null @@ -1,810 +0,0 @@ -import builtins -import collections.abc -import itertools -import os -import shutil -import uuid -import time -from pathlib import Path -from datetime import datetime, timezone - -import dask.dataframe -import numpy -import pymongo -import sparse -import zarr -import zarr.storage - -from tiled.adapters.array import slice_and_shape_from_block_and_chunks -from tiled.adapters.array import ArrayAdapter -from tiled.adapters.dataframe import DataFrameAdapter -from tiled.adapters.mapping import MapAdapter -from tiled.adapters.utils import IndexersMixin, tree_repr -from tiled.iterviews import KeysView, ItemsView, ValuesView -from tiled.queries import ( - Contains, - Comparison, - Eq, - FullText, - In, - KeysFilter, - NotEq, - NotIn, - Operator, - Regex, -) -from tiled.query_registration import QueryTranslationRegistry - -from tiled.structures.core import Spec, StructureFamily -from tiled.structures.array import ArrayStructure, ArrayMacroStructure, BuiltinDtype -from tiled.structures.dataframe import ( - DataFrameStructure, - DataFrameMacroStructure, - DataFrameMicroStructure, -) -from tiled.structures.sparse import COOStructure -from tiled.serialization.dataframe import deserialize_arrow - -from tiled.utils import APACHE_ARROW_FILE_MIME_TYPE, UNCHANGED, OneShotCachedMap - -from .schemas import Document, DocumentRevision - -from sys import modules, platform - - -class Revisions: - def __init__(self, collection, key): - self._collection = collection - self._key = key - - def __len__(self): - return self._collection.count_documents({"key": self._key}) - - def __getitem__(self, item_): - offset = item_.start - limit = item_.stop - offset - return list( - self._collection.find({"key": self._key}) - .sort("revision", 1) - .skip(offset) - .limit(limit) - ) - - def delete_revision(self, n): - self._collection.delete_one({"key": self._key, "revision": n}) - - def delete_all(self): - self._collection.delete_many({"key": self._key}) - - def last(self): - return self._collection.find_one({"key": self._key}, sort=[("revision", -1)]) - - def add_document(self, document): - return self._collection.insert_one(document.dict()) - - -class WritingArrayAdapter: - structure_family = "array" - - def __init__(self, database, key): - self.collection = database["nodes"] - self.revisions = Revisions(database["revisions"], key) - self.key = key - self.deadline = 0 - assert self.doc.data_blob is None # not implemented - self.array = zarr.open_array(str(safe_path(self.doc.data_url.path)), "r+") - - @classmethod - def new(cls, database, doc): - # Zarr requires evently-sized chunks within each dimension. - # Use the first chunk along each dimension. - chunks = tuple(dim[0] for dim in doc.structure.macro.chunks) - shape = tuple(dim[0] * len(dim) for dim in doc.structure.macro.chunks) - storage = zarr.storage.DirectoryStore(str(safe_path(doc.data_url.path))) - zarr.storage.init_array( - storage, - shape=shape, - chunks=chunks, - dtype=doc.structure.micro.to_numpy_dtype(), - ) - return cls(database, doc.key) - - @property - def doc(self): - now = time.monotonic() - if now > self.deadline: - self._doc = Document(**self.collection.find_one({"key": self.key})) - self.deadline = now + 0.1 # In seconds - - return self._doc - - @property - def structure(self): - # Convert pydantic implementation to dataclass implemenetation - # expected by server. - return ArrayStructure(**self.doc.structure.dict()) - - @property - def metadata(self): - return self.doc.metadata - - @property - def specs(self): - # Convert from pydantic to dataclass. - return [Spec(**dict(spec)) for spec in self.doc.specs] - - @property - def references(self): - return self.doc.references - - def read(self, slice=None): - # Trim overflow because Zarr always has equal-sized chunks. - arr = self.array[ - tuple(builtins.slice(0, dim) for dim in self.doc.structure.macro.shape) - ] - if slice is not None: - arr = arr[slice] - return arr - - def read_block(self, block, slice=None): - # Trim overflow because Zarr always has equal-sized chunks. - slice_, _ = slice_and_shape_from_block_and_chunks( - block, self.doc.structure.macro.chunks - ) - # Slice the block out of the whole array. - arr = self.array[slice_] - # And then maybe slice *within* the block. - if slice is not None: - arr = arr[slice] - return arr - - def microstructure(self): - return BuiltinDtype(**self.doc.structure.micro.dict()) - - def macrostructure(self): - return ArrayMacroStructure(**self.doc.structure.macro.dict()) - - def put_data(self, body, block=None): - # Organize files into subdirectories with the first two - # characters of the key to avoid one giant directory. - if block: - slice_, shape = slice_and_shape_from_block_and_chunks( - block, self.doc.structure.macro.chunks - ) - else: - slice_ = numpy.s_[:] - shape = self.doc.structure.macro.shape - array = numpy.frombuffer( - body, dtype=self.doc.structure.micro.to_numpy_dtype() - ).reshape(shape) - self.array[slice_] = array - - def put_metadata(self, metadata, specs, references): - last_revision_doc = self.revisions.last() - if last_revision_doc is not None: - revision = int(last_revision_doc["revision"]) + 1 - else: - revision = 0 - - validated_revision = DocumentRevision.from_document(self.doc, revision) - - self.revisions.add_document(validated_revision) - updated_at = datetime.now(tz=timezone.utc) - - to_set = {"updated_at": updated_at} - - if metadata is not None: - to_set["metadata"] = metadata - - if specs is not None: - to_set["specs"] = [spec.dict() for spec in specs] - - if references is not None: - references_dict = [item.dict() for item in references] - to_set["references"] = references_dict - - result = self.collection.update_one( - {"key": self.key}, - {"$set": to_set}, - ) - - if result.matched_count != result.modified_count: - raise RuntimeError("Error while writing to database") - - def delete(self): - shutil.rmtree(safe_path(self.doc.data_url.path)) - result = self.collection.delete_one({"key": self.key}) - assert result.deleted_count == 1 - self.revisions.delete_all() - - -class WritingDataFrameAdapter: - structure_family = "dataframe" - - def __init__(self, database, key): - self.collection = database["nodes"] - self.revisions = Revisions(database["revisions"], key) - self.key = key - self.deadline = 0 - assert self.doc.data_blob is None # not implemented - - @property - def dataframe_adapter(self): - return DataFrameAdapter.from_dask_dataframe( - dask.dataframe.read_parquet(safe_path(self.doc.data_url.path)) - ) - - @classmethod - def new(cls, database, doc): - safe_path(doc.data_url.path).mkdir(parents=True) - return cls(database, doc.key) - - @property - def doc(self): - now = time.monotonic() - if now > self.deadline: - self._doc = Document( - **self.collection.find_one({"key": self.key}) - ) # run query - self.deadline = now + 0.1 # In seconds - - return self._doc - - @property - def structure(self): - # Convert pydantic implementation to dataclass implemenetation - # expected by server. - return DataFrameStructure(**self.doc.structure.dict()) - - @property - def metadata(self): - return self.doc.metadata - - @property - def specs(self): - # Convert from pydantic to dataclass. - return [Spec(**dict(spec)) for spec in self.doc.specs] - - @property - def references(self): - return self.doc.references - - def __getitem__(self, key): - return ArrayAdapter(self.dataframe_adapter.read([key])[key].values) - - def read(self, *args, **kwargs): - return self.dataframe_adapter.read(*args, **kwargs) - - def read_partition(self, *args, **kwargs): - return self.dataframe_adapter.read_partition(*args, **kwargs) - - def microstructure(self): - return DataFrameMicroStructure(**self.doc.structure.micro.dict()) - - def macrostructure(self): - return DataFrameMacroStructure(**self.doc.structure.macro.dict()) - - def put_data(self, body, partition=0): - dataframe = deserialize_arrow(body) - dataframe.to_parquet( - safe_path(self.doc.data_url.path) / f"partition-{partition}.parquet" - ) - - def put_metadata(self, metadata, specs, references): - last_revision_doc = self.revisions.last() - if last_revision_doc is not None: - revision = int(last_revision_doc["revision"]) + 1 - else: - revision = 0 - - validated_revision = DocumentRevision.from_document(self.doc, revision) - - result = self.revisions.add_document(validated_revision) - updated_at = datetime.now(tz=timezone.utc) - - to_set = {"updated_at": updated_at} - - if metadata is not None: - to_set["metadata"] = metadata - - if specs is not None: - to_set["specs"] = [spec.dict() for spec in specs] - - if references is not None: - references_dict = [item.dict() for item in references] - to_set["references"] = references_dict - - result = self.collection.update_one( - {"key": self.key}, - {"$set": to_set}, - ) - - if result.matched_count != result.modified_count: - raise RuntimeError("Error while writing to database") - - def delete(self): - shutil.rmtree(safe_path(self.doc.data_url.path)) - result = self.collection.delete_one({"key": self.doc.key}) - assert result.deleted_count == 1 - self.revisions.delete_all() - - -class WritingCOOAdapter: - structure_family = "sparse" - - def __init__(self, database, key): - def load(filepath): - import pandas - - df = pandas.read_parquet(filepath) - coords = df[df.columns[:-1]].values.T - data = df["data"].values - return coords, data - - self.collection = database["nodes"] - self.revisions = Revisions(database["revisions"], key) - self.key = key - self.deadline = 0 - assert self.doc.data_blob is None # not implemented - num_blocks = (range(len(n)) for n in self.doc.structure.chunks) - directory = safe_path(self.doc.data_url.path) - mapping = {} - for block in itertools.product(*num_blocks): - filepath = directory / f"block-{'.'.join(map(str, block))}.parquet" - if filepath.is_file(): - mapping[block] = lambda filepath=filepath: load(filepath) - self.blocks = OneShotCachedMap(mapping) - - @property - def doc(self): - now = time.monotonic() - if now > self.deadline: - self._doc = Document( - **self.collection.find_one({"key": self.key}) - ) # run query - self.deadline = now + 0.1 # In seconds - - return self._doc - - @classmethod - def new(cls, database, doc): - safe_path(doc.data_url.path).mkdir(parents=True) - return cls(database, doc.key) - - @property - def metadata(self): - return self.doc.metadata - - @property - def specs(self): - # Convert from pydantic to dataclass. - return [Spec(**dict(spec)) for spec in self.doc.specs] - - @property - def references(self): - return self.doc.references - - def read_block(self, block, slice=None): - coords, data = self.blocks[block] - _, shape = slice_and_shape_from_block_and_chunks( - block, self.doc.structure.chunks - ) - arr = sparse.COO(data=data[:], coords=coords[:], shape=shape) - if slice: - arr = arr[slice] - return arr - - def read(self, slice=None): - all_coords = [] - all_data = [] - for block, (coords, data) in self.blocks.items(): - offsets = [] - for b, c in zip(block, self.doc.structure.chunks): - offset = sum(c[:b]) - offsets.append(offset) - global_coords = coords + [[i] for i in offsets] - all_coords.append(global_coords) - all_data.append(data) - arr = sparse.COO( - data=numpy.concatenate(all_data), - coords=numpy.concatenate(all_coords, axis=-1), - shape=self.doc.structure.shape, - ) - if slice: - return arr[slice] - return arr - - def structure(self): - # Convert pydantic implementation to dataclass implemenetation - # expected by server. - return COOStructure(**self.doc.structure.dict()) - - def put_data(self, body, block=None): - if block is None: - block = (0,) * len(self.doc.structure.shape) - dataframe = deserialize_arrow(body) - dataframe.to_parquet( - safe_path(self.doc.data_url.path) - / f"block-{'.'.join(map(str, block))}.parquet" - ) - - def put_metadata(self, metadata, specs, references): - last_revision_doc = self.revisions.last() - if last_revision_doc is not None: - revision = int(last_revision_doc["revision"]) + 1 - else: - revision = 0 - - validated_revision = DocumentRevision.from_document(self.doc, revision) - - result = self.revisions.add_document(validated_revision) - updated_at = datetime.now(tz=timezone.utc) - - to_set = {"updated_at": updated_at} - - if metadata is not None: - to_set["metadata"] = metadata - - if specs is not None: - to_set["specs"] = [spec.dict() for spec in specs] - - if references is not None: - references_dict = [item.dict() for item in references] - to_set["references"] = references_dict - - result = self.collection.update_one( - {"key": self.key}, - {"$set": to_set}, - ) - - if result.matched_count != result.modified_count: - raise RuntimeError("Error while writing to database") - - def delete(self): - shutil.rmtree(safe_path(self.doc.data_url.path)) - result = self.collection.delete_one({"key": self.doc.key}) - assert result.deleted_count == 1 - self.revisions.delete_all() - - -class MongoAdapter(collections.abc.Mapping, IndexersMixin): - structure_family = "node" - query_registry = QueryTranslationRegistry() - register_query = query_registry.register - - def __init__( - self, - *, - database, - directory, - queries=None, - sorting=None, - metadata=None, - access_policy=None, - ): - self.database = database - self.collection = database["nodes"] - self.revision_coll = database["revisions"] - self.directory = Path(directory).resolve() - if not self.directory.exists(): - raise ValueError(f"Directory {self.directory} does not exist.") - if not self.directory.is_dir(): - raise ValueError( - f"The given directory path {self.directory} is not a directory." - ) - if not os.access(self.directory, os.W_OK): - raise ValueError(f"Directory {self.directory} is not writeable.") - self.queries = queries or [] - self.sorting = sorting or [("metadata.scan_id", 1)] - self.metadata = metadata or {} - self.access_policy = access_policy - super().__init__() - - @classmethod - def from_uri(cls, uri, directory, **kwargs): - """ - When calling this method, call create_index() from its instance to define the - unique indexes in the revision collection - - """ - if not pymongo.uri_parser.parse_uri(uri)["database"]: - raise ValueError( - f"Invalid URI: {uri!r} Did you forget to include a database?" - ) - client = pymongo.MongoClient(uri) - database = client.get_database() - return cls(database=database, directory=directory, **kwargs) - - @classmethod - def from_mongomock(cls, directory, **kwargs): - import mongomock - - db_name = f"temp-{str(uuid.uuid4())}" - mongo_client = mongomock.MongoClient() - database = mongo_client[db_name] - - mongo_adapter = cls(database=database, directory=directory, **kwargs) - mongo_adapter.create_indexes() - - return mongo_adapter - - def new_variation( - self, - metadata=UNCHANGED, - queries=UNCHANGED, - sorting=UNCHANGED, - **kwargs, - ): - if metadata is UNCHANGED: - metadata = self.metadata - if queries is UNCHANGED: - queries = self.queries - if sorting is UNCHANGED: - sorting = self.sorting - return type(self)( - database=self.database, - directory=self.directory, - metadata=metadata, - queries=queries, - sorting=sorting, - access_policy=self.access_policy, - **kwargs, - ) - - def post_metadata(self, metadata, structure_family, structure, specs, references): - mime_structure_association = { - StructureFamily.array: "application/x-zarr", - StructureFamily.dataframe: APACHE_ARROW_FILE_MIME_TYPE, - StructureFamily.sparse: APACHE_ARROW_FILE_MIME_TYPE, - } - - key = str(uuid.uuid4()) - created_date = datetime.now(tz=timezone.utc) - - validated_document = Document( - key=key, - structure_family=structure_family, - structure=structure, - metadata=metadata, - specs=specs, - references=references, - mimetype=mime_structure_association[structure_family], - data_url=f"file://localhost/{self.directory}/{key[:2]}/{key}", - created_at=created_date, - updated_at=created_date, - ) - - _adapter_class_by_family[structure_family] - self.collection.insert_one(validated_document.dict()) - _adapter_class_by_family[structure_family].new( - self.database, validated_document - ) - return key - - def create_indexes(self): - self.collection.create_index([("key", pymongo.ASCENDING)], unique=True) - - self.revision_coll.create_index( - [("key", pymongo.ASCENDING), ("revision", pymongo.DESCENDING)], unique=True - ) - - def _build_mongo_query(self, *queries): - combined = self.queries + list(queries) - if combined: - return {"$and": combined} - else: - return {} - - def __getitem__(self, key): - query = {"key": key} - doc = self.collection.find_one(self._build_mongo_query(query), {"_id": False}) - if doc is None: - raise KeyError(key) - - class_ = _adapter_class_by_family[StructureFamily(doc["structure_family"])] - return class_(self.database, key) - - def __iter__(self): - # TODO Apply pagination, as we do in Databroker. - for doc in list( - self.collection.find( - self._build_mongo_query({"data_url": {"$ne": None}}), - {"key": True}, - ) - ): - yield doc["key"] - - def __len__(self): - return self.collection.count_documents( - self._build_mongo_query({"data_url": {"$ne": None}}) - ) - - def __length_hint__(self): - # https://www.python.org/dev/peps/pep-0424/ - return self.collection.estimated_document_count( - self._build_mongo_query({"data_url": {"$ne": None}}), - ) - - def __repr__(self): - # Display up to the first N keys to avoid making a giant service - # request. Use _keys_slicer because it is unauthenticated. - N = 10 - return tree_repr(self, self._keys_slice(0, N, direction=1)) - - def search(self, query): - """ - Return a MongoAdapter with a subset of the mapping. - """ - return self.query_registry(query, self) - - def get_distinct(self, metadata, structure_families, specs, counts): - data = {} - - select = {"$match": self._build_mongo_query({"data_url": {"$ne": None}})} - - if counts: - project = {"$project": {"_id": 0, "value": "$_id", "count": "$count"}} - else: - project = {"$project": {"_id": 0, "value": "$_id"}} - - if metadata: - data["metadata"] = {} - for metadata_key in metadata: - group = { - "$group": {"_id": f"$metadata.{metadata_key}", "count": {"$sum": 1}} - } - data["metadata"][f"{metadata_key}"] = list( - self.collection.aggregate([select, group, project]) - ) - - if structure_families: - group = {"$group": {"_id": "$structure_family", "count": {"$sum": 1}}} - data["structure_families"] = list( - self.collection.aggregate([select, group, project]) - ) - - if specs: - group = {"$group": {"_id": "$specs", "count": {"$sum": 1}}} - distinct_list = list(self.collection.aggregate([select, group, project])) - data["specs"] = distinct_list - - return data - - def sort(self, sorting): - return self.new_variation(sorting=sorting) - - def keys(self): - return KeysView(lambda: len(self), self._keys_slice) - - def values(self): - return ValuesView(lambda: len(self), self._items_slice) - - def items(self): - return ItemsView(lambda: len(self), self._items_slice) - - def _keys_slice(self, start, stop, direction): - assert direction == 1, "direction=-1 should be handled by the client" - skip = start or 0 - if stop is not None: - limit = stop - skip - else: - limit = None - for doc in self.collection.find( - self._build_mongo_query({"data_url": {"$ne": None}}), - skip=skip, - limit=limit, - ): - yield doc["key"] - - def _items_slice(self, start, stop, direction): - assert direction == 1, "direction=-1 should be handled by the client" - skip = start or 0 - if stop is not None: - limit = stop - skip - else: - limit = None - - for doc in self.collection.find( - self._build_mongo_query({"data_url": {"$ne": None}}), - skip=skip, - limit=limit, - ): - class_ = _adapter_class_by_family[StructureFamily(doc["structure_family"])] - yield doc["key"], class_(self.database, doc["key"]) - - def apply_mongo_query(self, query): - return self.new_variation( - queries=self.queries + [query], - ) - - -def contains(query, catalog): - # In MongoDB, checking that an item is in an array looks - # just like equality. - # https://www.mongodb.com/docs/manual/tutorial/query-arrays/ - return catalog.apply_mongo_query({f"metadata.{query.key}": query.value}) - - -def comparison(query, catalog): - OPERATORS = { - Operator.lt: "$lt", - Operator.le: "$lte", - Operator.gt: "$gt", - Operator.ge: "$gte", - } - return catalog.apply_mongo_query( - {f"metadata.{query.key}": {OPERATORS[query.operator]: query.value}} - ) - - -def eq(query, catalog): - return catalog.apply_mongo_query({f"metadata.{query.key}": query.value}) - - -def noteq(query, catalog): - return catalog.apply_mongo_query({f"metadata.{query.key}": {"$ne": query.value}}) - - -def regex(query, catalog): - options = "" if query.case_sensitive else "i" - return catalog.apply_mongo_query( - {f"metadata.{query.key}": {"$regex": query.pattern, "$options": options}} - ) - - -def _in(query, catalog): - if not isinstance(query.value, list): - query.value = [query.value] - return catalog.apply_mongo_query({f"metadata.{query.key}": {"$in": query.value}}) - - -def notin(query, catalog): - if not isinstance(query.value, list): - query.value = [query.value] - return catalog.apply_mongo_query({f"metadata.{query.key}": {"$nin": query.value}}) - - -def full_text_search(query, catalog): - # First if this catalog is backed by mongomock, which does not support $text queries. - # Avoid importing mongomock if it is not already imported. - if "mongomock" in modules: - import mongomock - - if isinstance(catalog.database.client, mongomock.MongoClient): - # Do the query in memory. - # For huge MongoAdapters this will be slow, but if you are attempting - # full text search on a large mongomock-backed MongoAdapter, - # you have made your choices! :-) - return MapAdapter(dict(catalog)).search(query) - - return catalog.apply_mongo_query( - {"$text": {"$search": query.text, "$caseSensitive": query.case_sensitive}}, - ) - - -def keys_filter(query, catalog): - return catalog.apply_mongo_query({"key": {"$in": list(query.keys)}}) - - -MongoAdapter.register_query(Contains, contains) -MongoAdapter.register_query(Comparison, comparison) -MongoAdapter.register_query(Eq, eq) -MongoAdapter.register_query(NotEq, noteq) -MongoAdapter.register_query(Regex, regex) -MongoAdapter.register_query(In, _in) -MongoAdapter.register_query(KeysFilter, keys_filter) -MongoAdapter.register_query(NotIn, notin) -MongoAdapter.register_query(FullText, full_text_search) - - -def safe_path(path): - if platform == "win32" and path[0] == "/": - path = path[1:] - return Path(path) - - -_adapter_class_by_family = { - StructureFamily.array: WritingArrayAdapter, - StructureFamily.dataframe: WritingDataFrameAdapter, - StructureFamily.sparse: WritingCOOAdapter, -} diff --git a/databroker/tests/test_experimental.py b/databroker/tests/test_experimental.py deleted file mode 100644 index ac5baa5bb..000000000 --- a/databroker/tests/test_experimental.py +++ /dev/null @@ -1,558 +0,0 @@ -import contextlib -import string - -import dask.array -import dask.dataframe -import httpx -import numpy -import pandas -import pytest -import sparse -from tiled.client import Context, from_context -from tiled.queries import ( - Contains, - Comparison, - Eq, - FullText, - In, - Key, - NotEq, - NotIn, - Regex, -) -from tiled.server.app import build_app, build_app_from_config -from tiled.structures.core import Spec -from tiled.structures.sparse import COOStructure -from tiled.validation_registration import ValidationRegistry - -from ..experimental.server_ext import MongoAdapter -from ..experimental.schemas import DocumentRevision - -# from .test_access_policy import enter_password - - -API_KEY = "secret" -validation_registry = ValidationRegistry() -validation_registry.register("SomeSpec", lambda *args, **kwargs: None) -validation_registry.register("AnotherSpec", lambda *args, **kwargs: None) -validation_registry.register("AnotherOtherSpec", lambda *args, **kwargs: None) - - -@pytest.fixture -def client(tmpdir): - tree = MongoAdapter.from_mongomock(tmpdir) - app = build_app(tree, validation_registry=validation_registry) - with Context.from_app(app) as context: - client = from_context(context) - yield client - - -def test_write_array(client): - test_array = numpy.ones((5, 7)) - - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - - node = client.write_array( - test_array, metadata=metadata, specs=specs, references=references - ) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_array = result.read() - - numpy.testing.assert_equal(result_array, test_array) - assert result.metadata == node.metadata == metadata - assert result.specs == node.specs == specs - assert result.references == node.references == references - - -def test_write_dataframe(client): - dummy_array = numpy.ones((5, 7)) - - data = { - "Column1": dummy_array[0], - "Column2": dummy_array[1], - "Column3": dummy_array[2], - "Column4": dummy_array[3], - "Column5": dummy_array[4], - } - - test_dataframe = pandas.DataFrame(data) - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - - node = client.write_dataframe( - test_dataframe, metadata=metadata, specs=specs, references=references - ) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_dataframe = result.read() - - pandas.testing.assert_frame_equal(result_dataframe, test_dataframe) - # slicing into DataFrameClient returns ArrayClient - result_array = result["Column1"][:] - assert numpy.array_equal(result_array, dummy_array[0]) - assert result.metadata == node.metadata == metadata - assert result.specs == node.specs == specs - assert result.references == node.references == references - - -def test_queries(client): - keys = list(string.ascii_lowercase) - - for letter, number in zip(keys, range(26)): - metadata = {"letter": letter, "number": number} - array = number * numpy.ones(10) - - client.write_array(array, metadata=metadata) - - test1 = client.search(Eq("letter", "a")) - numpy.testing.assert_equal( - test1.values()[0].read(), test1.values()[0].metadata["number"] * numpy.ones(10) - ) - test2 = client.search(Contains("number", 1)) - numpy.testing.assert_equal( - test2.values()[0].read(), test2.values()[0].metadata["number"] * numpy.ones(10) - ) - test3 = client.search(Comparison("gt", "number", 24)) - numpy.testing.assert_equal( - test3.values()[0].read(), test3.values()[0].metadata["number"] * numpy.ones(10) - ) - test4 = client.search(FullText("y")) - numpy.testing.assert_equal( - test4.values()[0].read(), test4.values()[0].metadata["number"] * numpy.ones(10) - ) - test5 = client.search(Regex("letter", "^c$")) - numpy.testing.assert_equal( - test5.values()[0].read(), test5.values()[0].metadata["number"] * numpy.ones(10) - ) - test6 = client.search(NotEq("letter", "a")) - # The first result should not be "a" - assert test6.values()[0].metadata["letter"] != "a" - - test7 = client.search(In("letter", ["a", "b"])) - numpy.testing.assert_equal( - test7.values()[0].read(), test7.values()[0].metadata["number"] * numpy.ones(10) - ) - numpy.testing.assert_equal( - test7.values()[1].read(), test7.values()[1].metadata["number"] * numpy.ones(10) - ) - - test8 = client.search(NotIn("letter", ["a"])) - # The first result should not be "a" - assert test8.values()[0].metadata["letter"] != "a" - - -def test_delete(client): - # For dataframes - dummy_array = numpy.ones((5, 5)) - - data = { - "Column1": dummy_array[0], - "Column2": dummy_array[1], - "Column3": dummy_array[2], - "Column4": dummy_array[3], - "Column5": dummy_array[4], - } - - test_dataframe = pandas.DataFrame(data) - - x = client.write_dataframe( - test_dataframe, - metadata={"scan_id": 1, "method": "A"}, - specs=["SomeSpec"], - references=[{"label": "test", "url": "http://www.test.com"}], - ) - - del client[x.item["id"]] - - assert x.item["id"] not in client - - # For arrays - test_array = numpy.ones((5, 5)) - - y = client.write_array( - test_array, - metadata={"scan_id": 1, "method": "A"}, - specs=["SomeSpec"], - references=[{"label": "test", "url": "http://www.test.com"}], - ) - - del client[y.item["id"]] - - assert y.item["id"] not in client - - -def test_write_array_chunked(client): - a = dask.array.arange(24).reshape((4, 6)).rechunk((2, 3)) - - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - client.write_array(a, metadata=metadata, specs=specs, references=references) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_array = result.read() - - numpy.testing.assert_equal(result_array, a.compute()) - assert result.metadata == metadata - assert result.specs == specs - assert result.references == references - - -def test_write_dataframe_partitioned(client): - data = {f"Column{i}": (1 + i) * numpy.ones(10) for i in range(5)} - df = pandas.DataFrame(data) - ddf = dask.dataframe.from_pandas(df, npartitions=3) - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - - client.write_dataframe(ddf, metadata=metadata, specs=specs, references=references) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_dataframe = result.read() - - pandas.testing.assert_frame_equal(result_dataframe, df) - assert result.metadata == metadata - # TODO In the future this will be accessible via result.specs. - assert result.specs == specs - assert result.references == references - - -def test_write_sparse_full(client): - coo = sparse.COO(coords=[[0, 1], [2, 3]], data=[3.8, 4.0], shape=(4, 4)) - - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - client.write_sparse( - coords=coo.coords, - data=coo.data, - shape=coo.shape, - metadata=metadata, - specs=specs, - references=references, - ) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_array = result.read() - - numpy.testing.assert_equal(result_array.todense(), coo.todense()) - assert result.metadata == metadata - assert result.specs == specs - assert result.references == references - - -def test_write_sparse_chunked(client): - metadata = {"scan_id": 1, "method": "A"} - specs = [Spec("SomeSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - N = 5 - x = client.new( - "sparse", - COOStructure(shape=(2 * N,), chunks=((N, N),)), - metadata=metadata, - specs=specs, - references=references, - ) - x.write_block(coords=[[2, 4]], data=[3.1, 2.8], block=(0,)) - x.write_block(coords=[[0, 1]], data=[6.7, 1.2], block=(1,)) - - results = client.search(Key("scan_id") == 1) - result = results.values().first() - result_array = result.read() - assert numpy.array_equal( - result_array.todense(), - sparse.COO( - coords=[[2, 4, N + 0, N + 1]], data=[3.1, 2.8, 6.7, 1.2], shape=(10,) - ).todense(), - ) - - # numpy.testing.assert_equal(result_array, sparse.COO(coords=[0, 1, ])) - assert result.metadata == metadata - assert result.specs == specs - assert result.references == references - - -def test_update_array_metadata(client): - # Update metadata in array client - test_array = numpy.ones((5, 5)) - - x = client.write_array( - test_array, metadata={"scan_id": 1, "method": "A"}, specs=["SomeSpec"] - ) - - new_arr_metadata = {"scan_id": 2, "method": "A"} - new_spec = [Spec("AnotherSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - x.update_metadata(new_arr_metadata, new_spec, references) - - # validate local data after update request - assert x.metadata == new_arr_metadata - assert x.specs == new_spec - assert x.references == references - - # Update metadata again to create another entry in revisions - newer_arr_metadata = {"scan_id": 2, "method": "B"} - newer_spec = [Spec("AnotherOtherSpec")] - new_references = [{"label": "updated_test", "url": "http://www.updatedtest.com"}] - x.update_metadata(newer_arr_metadata, newer_spec, new_references) - - # Increase the size of revisions for additonal testing - latest_arr_metadata = {"scan_id": 2, "method": "C"} - x.update_metadata(latest_arr_metadata) - - results = client.search(Key("scan_id") == 2) - result = results.values().first() - - # validate remote data after update request - assert result.metadata == latest_arr_metadata - assert result.specs == newer_spec - assert result.references == new_references - - rev_document = { - "key": result.item["id"], - "revision": result.metadata_revisions[0]["revision"], - } - rev_document.update(result.metadata_revisions[0]["attributes"]) - assert DocumentRevision.from_json(rev_document) - - assert len(result.metadata_revisions[0:2]) == 2 - assert len(result.metadata_revisions) == len(result.metadata_revisions[:]) - - result.metadata_revisions.delete_revision(0) - assert len(result.metadata_revisions[:]) == 2 - - -def test_update_dataframe_metadata(client): - test_array = numpy.ones((5, 5)) - - # Update metadata in dataframe client - data = { - "Column1": test_array[0], - "Column2": test_array[1], - "Column3": test_array[2], - "Column4": test_array[3], - "Column5": test_array[4], - } - - test_dataframe = pandas.DataFrame(data) - - y = client.write_dataframe( - test_dataframe, metadata={"scan_id": 1, "method": "A"}, specs=["SomeSpec"] - ) - - new_df_metadata = {"scan_id": 2, "method": "A"} - new_spec = [Spec("AnotherSpec")] - references = [{"label": "test", "url": "http://www.test.com"}] - y.update_metadata(new_df_metadata, new_spec, references) - - # validate local data after update request - assert y.metadata == new_df_metadata - assert y.specs == new_spec - assert y.references == references - - # Update metadata again to create another entry in revisions - newer_df_metadata = {"scan_id": 2, "method": "B"} - newer_spec = [Spec("AnotherOtherSpec")] - new_references = [{"label": "updated_test", "url": "http://www.updatedtest.com"}] - y.update_metadata(newer_df_metadata, newer_spec, new_references) - - # Increase the size of revisions for additonal testing - latest_arr_metadata = {"scan_id": 2, "method": "C"} - y.update_metadata(latest_arr_metadata) - - results = client.search(Key("scan_id") == 2) - result = results.values().first() - - # validate remote data after update request - assert result.metadata == latest_arr_metadata - assert result.specs == newer_spec - assert result.references == new_references - - rev_document = { - "key": result.item["id"], - "revision": result.metadata_revisions[0]["revision"], - } - rev_document.update(result.metadata_revisions[0]["attributes"]) - assert DocumentRevision.from_json(rev_document) - - assert len(result.metadata_revisions[0:2]) == 2 - assert len(result.metadata_revisions) == len(result.metadata_revisions[:]) - - result.metadata_revisions.delete_revision(0) - assert len(result.metadata_revisions[:]) == 2 - - -@contextlib.contextmanager -def fail_with_status_code(status_code): - with pytest.raises(httpx.HTTPStatusError) as info: - yield - assert info.value.response.status_code == status_code - - -def test_simple_access_policy(tmpdir, enter_password): - config = { - "authentication": { - "providers": [ - { - "provider": "toy", - "authenticator": "tiled.authenticators:DictionaryAuthenticator", - "args": { - "users_to_passwords": { - "alice": "secret1", - "bob": "secret2", - "cara": "secret3", - } - }, - } - ], - }, - "trees": [ - { - "path": "/", - "tree": "databroker.experimental.server_ext:MongoAdapter.from_mongomock", - "args": {"directory": tmpdir}, - "access_control": { - "access_policy": "tiled.access_policies:SimpleAccessPolicy", - "args": { - "provider": "toy", - "access_lists": { - "alice": [], - "cara": "tiled.access_policies:ALL_ACCESS", - }, - }, - }, - } - ], - } - with Context.from_app(build_app_from_config(config), token_cache=tmpdir) as context: - # User with all access - with enter_password("secret3"): - client = from_context( - context, username="cara", prompt_for_reauthentication=True - ) - client.write_array([1, 2, 3]) - assert len(list(client)) == 1 - # User with no access - with enter_password("secret1"): - client = from_context( - context, username="alice", prompt_for_reauthentication=True - ) - assert len(list(client)) == 0 - # User with implicitly no access (not mentioned in policy) - with enter_password("secret2"): - client = from_context( - context, username="bob", prompt_for_reauthentication=True - ) - assert len(list(client)) == 0 - - -def test_distinct(client): - # Begin of test data generation - - # Added additional field in metadata to implement consecutive search and distinct queries - for i in range(10): - if i < 5: - group = "A" - else: - group = "B" - - if i % 2 == 0: - subgroup = "even" - specs = [Spec("test")] - else: - subgroup = "odd" - specs = [Spec("test"), Spec("moreTest")] - - if i == 0: - tag = "Zero" - else: - for j in range(2, int(i / 2) + 1): - if (i % j) == 0: - tag = "NotPrime" - break - else: - tag = "Prime" - - df = pandas.DataFrame({"a": i * numpy.ones(10)}) - metadata = {"group": group, "subgroup": subgroup, "tag": tag} - - client.write_dataframe(df, metadata=metadata, specs=specs) - - # End of test data generation - - # Not Counting - results = client.search(Key("group") == "B").distinct( - "tag", structure_families=True, specs=True, counts=False - ) - - # Results are retrieved from the database as an unsorted list. - # They are sorted by count to validate them during the test run. - results["metadata"]["tag"].sort(key=lambda k: k["value"]) - results["specs"].sort(key=lambda k: k["value"]) - - expected = { - "metadata": { - "tag": [ - {"value": "NotPrime", "count": None}, - {"value": "Prime", "count": None}, - ] - }, - "specs": [ - {"value": [{"name": "test", "version": None}], "count": None}, - { - "value": [ - {"name": "test", "version": None}, - {"name": "moreTest", "version": None}, - ], - "count": None, - }, - ], - "structure_families": [{"value": "dataframe", "count": None}], - } - - assert results["metadata"] == expected["metadata"] - assert results["specs"] == expected["specs"] - assert results["structure_families"] == expected["structure_families"] - - # Counting - results = client.search(Key("group") == "B").distinct( - "tag", structure_families=True, specs=True, counts=True - ) - - # Results are retrieved from the database as an unsorted list. - # They are sorted by count to validate them during the test run. - results["metadata"]["tag"].sort(key=lambda k: k["count"]) - results["specs"].sort(key=lambda k: k["count"]) - - expected = { - "metadata": { - "tag": [{"value": "Prime", "count": 2}, {"value": "NotPrime", "count": 3}] - }, - "specs": [ - {"value": [{"name": "test", "version": None}], "count": 2}, - { - "value": [ - {"name": "test", "version": None}, - {"name": "moreTest", "version": None}, - ], - "count": 3, - }, - ], - "structure_families": [ - {"value": "dataframe", "count": 5}, - ], - } - - assert results["metadata"] == expected["metadata"] - assert results["specs"] == expected["specs"] - assert results["structure_families"] == expected["structure_families"] From 4476f1c780753c07c641fc403bbbed9d72d75fac Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 09:30:08 -0400 Subject: [PATCH 2/7] Adjust test for change in behavior due to pre-declared streams. --- databroker/tests/test_broker.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/databroker/tests/test_broker.py b/databroker/tests/test_broker.py index f239bb3bc..55e4724fb 100644 --- a/databroker/tests/test_broker.py +++ b/databroker/tests/test_broker.py @@ -26,7 +26,7 @@ if sys.version_info >= (3, 5): from bluesky.plans import count - from bluesky.plan_stubs import trigger_and_read, configure + from bluesky.plan_stubs import trigger_and_read, configure, one_shot from bluesky.preprocessors import (monitor_during_wrapper, run_decorator, baseline_wrapper, @@ -86,11 +86,20 @@ def test_uid_list_multiple_headers(db, RE, hw): def test_no_descriptors(db, RE): RE.subscribe(db.insert) - uid, = get_uids(RE(count([]))) + # Specifying per_shot as 'not None' turns off pre-declaring the stream and + # results in no descriptors. + uid, = get_uids(RE(count([], per_shot=one_shot), print)) header = db[uid] assert [] == header.descriptors +def test_no_events(db, RE): + RE.subscribe(db.insert) + uid, = get_uids(RE(count([]))) + header = db[uid] + header.table() + + def test_get_events(db, RE, hw): RE.subscribe(db.insert) uid, = get_uids(RE(count([hw.det]))) From a7b231253216b9b0065638f808b243688bc54c7c Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 09:35:21 -0400 Subject: [PATCH 3/7] Fix long-standing bug when stream has 0 events. --- databroker/mongo_normalized.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/databroker/mongo_normalized.py b/databroker/mongo_normalized.py index dd55b3b0f..cf111ceea 100644 --- a/databroker/mongo_normalized.py +++ b/databroker/mongo_normalized.py @@ -1432,10 +1432,14 @@ def _build_event_stream(self, *, run_start_uid, stream_name, is_complete): }, ] ) - (result,) = cursor - cutoff_seq_num = ( - 1 + result["highest_seq_num"] - ) # `1 +` because we use a half-open interval + results = list(cursor) + if results: + (result,) = results + cutoff_seq_num = ( + 1 + result["highest_seq_num"] + ) # `1 +` because we use a half-open interval + else: + cutoff_seq_num = 1 object_names = event_descriptors[0]["object_keys"] run = self[run_start_uid] mapping = OneShotCachedMap( From 02e3d3e0d2878e07098150eb7dee48bc0383543a Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 09:43:30 -0400 Subject: [PATCH 4/7] xfail a test for now --- databroker/tests/test_broker.py | 1 + 1 file changed, 1 insertion(+) diff --git a/databroker/tests/test_broker.py b/databroker/tests/test_broker.py index 55e4724fb..f73c46cc8 100644 --- a/databroker/tests/test_broker.py +++ b/databroker/tests/test_broker.py @@ -700,6 +700,7 @@ def test_dict_header(db, RE, hw): h['events'] +@pytest.mark.xfail(reason="Possibly broken by predeclare, need investigation") def test_config_data(db, RE, hw): # simple case: one Event Descriptor, one stream RE.subscribe(db.insert) From 539a2859f605c9708afe289fb6edba38c10ec496 Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 11:32:28 -0400 Subject: [PATCH 5/7] In CI matrix drop 3.7, add 3.10. --- .github/workflows/testing.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml index 0b9ba26d6..2163ef91d 100644 --- a/.github/workflows/testing.yml +++ b/.github/workflows/testing.yml @@ -16,7 +16,7 @@ jobs: - 27017:27017 strategy: matrix: - python-version: [3.7, 3.8, 3.9] + python-version: [3.8, 3.9, 3.10] fail-fast: false steps: From 378bb8c6e7ea97a59e6cafb84728714a4acb7c2c Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 11:54:24 -0400 Subject: [PATCH 6/7] Protect against testing against old bluesky. --- databroker/tests/test_broker.py | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/databroker/tests/test_broker.py b/databroker/tests/test_broker.py index f73c46cc8..7f691cd99 100644 --- a/databroker/tests/test_broker.py +++ b/databroker/tests/test_broker.py @@ -4,6 +4,7 @@ import tempfile import os import logging +import packaging import sys import string import time as ttime @@ -24,14 +25,14 @@ from databroker._core import DOCT_NAMES from databroker.tests.utils import get_uids -if sys.version_info >= (3, 5): - from bluesky.plans import count - from bluesky.plan_stubs import trigger_and_read, configure, one_shot - from bluesky.preprocessors import (monitor_during_wrapper, - run_decorator, - baseline_wrapper, - stage_wrapper, - pchain) +from bluesky import __version__ as bluesky_version +from bluesky.plans import count +from bluesky.plan_stubs import trigger_and_read, configure, one_shot +from bluesky.preprocessors import (monitor_during_wrapper, + run_decorator, + baseline_wrapper, + stage_wrapper, + pchain) logger = logging.getLogger(__name__) @@ -94,6 +95,8 @@ def test_no_descriptors(db, RE): def test_no_events(db, RE): + if packaging.version.parse(bluesky_version) < packaging.version.parse("1.11.0"): + pytest.skip("This test relies on the pre-declare streams feature added in bluesky 1.11") RE.subscribe(db.insert) uid, = get_uids(RE(count([]))) header = db[uid] From bbf62dfa546e533d98bbb80c0599c4deb705bf17 Mon Sep 17 00:00:00 2001 From: Dan Allan Date: Wed, 5 Jul 2023 12:00:45 -0400 Subject: [PATCH 7/7] Fix YAML error. --- .github/workflows/testing.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml index 2163ef91d..d559464c7 100644 --- a/.github/workflows/testing.yml +++ b/.github/workflows/testing.yml @@ -16,7 +16,7 @@ jobs: - 27017:27017 strategy: matrix: - python-version: [3.8, 3.9, 3.10] + python-version: ["3.8", "3.9", "3.10"] fail-fast: false steps: