Merge pull request #100 from AllenNeuralDynamics/release-v0.16.0

Release v0.16.0

.flake8

@@ -3,4 +3,4 @@ exclude =
     .git,
     __pycache__,
     build
-max-complexity = 10
+max-complexity = 10

pyproject.toml

@@ -20,7 +20,8 @@ dependencies = [
     "requests",
     "aind-codeocean-api>=0.4.0",
     "pydantic>=2.0",
-    "pydantic-settings>=2.0"
+    "pydantic-settings>=2.0",
+    "aind-data-schema",
 ]
 
 [project.optional-dependencies]
@@ -32,7 +33,7 @@ dev = [
     "isort",
     "Sphinx",
     "furo",
-    "aind-data-access-api[full]"
+    "aind-data-access-api[full]",
 ]
 secrets = [
     "boto3",

src/aind_data_access_api/__init__.py

@@ -1,3 +1,3 @@
 """Init package"""
 
-__version__ = "0.15.0"
+__version__ = "0.16.0"

src/aind_data_access_api/helpers/__init__.py

@@ -0,0 +1 @@
+"""Init module"""

src/aind_data_access_api/helpers/data_schema.py

@@ -0,0 +1,76 @@
+"""Module for convenience functions for the data access API."""
+
+from aind_data_access_api.document_db import MetadataDbClient
+from aind_data_access_api.helpers.docdb import (
+    get_field_by_id,
+    get_id_from_name,
+)
+from aind_data_schema.core.quality_control import QualityControl
+import json
+
+
+def get_quality_control_by_id(
+    client: MetadataDbClient,
+    _id: str,
+    allow_invalid: bool = False,
+):
+    """Using a connected DocumentDB client, retrieve the QualityControl object
+    for a given record.
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+        A connected DocumentDB client.
+    _id : str, optional
+        _id field in DocDB, by default empty
+    allow_invalid : bool, optional
+        return invalid QualityControl as dict if True, by default False
+    """
+    record = get_field_by_id(client, _id=_id, field="quality_control")
+    if not record:
+        raise ValueError(f"No record found with id {_id}")
+
+    if "quality_control" not in record or not record["quality_control"]:
+        raise ValueError(
+            f"No quality_control field found in record with id {_id}"
+        )
+
+    return validate_qc(record["quality_control"], allow_invalid=allow_invalid)
+
+
+def get_quality_control_by_name(
+    client: MetadataDbClient,
+    name: str,
+    allow_invalid: bool = False,
+):
+    """Using a connected DocumentDB client, retrieve the QualityControl object
+    for a given record.
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+        A connected DocumentDB client.
+    name : str, optional
+        name field in DocDB, by default empty
+    allow_invalid : bool, optional
+        return invalid QualityControl as dict if True, by default False
+    """
+    _id = get_id_from_name(client, name=name)
+    if not _id:
+        raise ValueError(f"No record found with name {name}")
+
+    return get_quality_control_by_id(
+        client, _id=_id, allow_invalid=allow_invalid
+    )
+
+
+def validate_qc(qc_data: dict, allow_invalid: bool = False):
+    """Validate a quality control dict."""
+
+    try:
+        return QualityControl.model_validate_json(json.dumps(qc_data))
+    except Exception as e:
+        if allow_invalid:
+            return qc_data
+        else:
+            raise e

src/aind_data_access_api/helpers/docdb.py

@@ -0,0 +1,114 @@
+"""Utilities that go through the MetadataDBClient """
+
+from typing import Optional
+from aind_data_access_api.document_db import MetadataDbClient
+import logging
+
+
+def get_record_by_id(
+    client: MetadataDbClient,
+    _id: str,
+) -> Optional[dict]:
+    """Download a record from docdb using the record _id.
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+    _id : str
+
+    Returns
+    -------
+    Optional[dict]
+        _description_
+    """
+    records = client.retrieve_docdb_records(filter_query={"_id": _id}, limit=1)
+    if len(records) > 0:
+        return records[0]
+    else:
+        return None
+
+
+def get_projection_by_id(
+    client: MetadataDbClient,
+    _id: str,
+    projection: dict,
+) -> Optional[dict]:
+    """
+    Download a record from docdb using the record _id and a projection.
+
+    Projections return fields set to 1 {"field": 1}
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+    _id : str
+    projection : dict
+
+    Returns
+    -------
+    Optional[dict]
+        None if record does not exist. Otherwise, it will return the projected
+        record as a dict.
+    """
+    records = client.retrieve_docdb_records(
+        filter_query={"_id": _id}, projection=projection, limit=1
+    )
+    if len(records) > 0:
+        return records[0]
+    else:
+        return None
+
+
+def get_field_by_id(
+    client: MetadataDbClient,
+    _id: str,
+    field: str,
+) -> Optional[dict]:
+    """Download a single field from docdb using the record _id
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+    _id : str
+    field : str
+
+    Returns
+    -------
+    Optional[dict]
+        None if a record does not exist. Otherwise returns the field in a dict.
+    """
+    return get_projection_by_id(client, _id=_id, projection={field: 1})
+
+
+def get_id_from_name(
+    client: MetadataDbClient,
+    name: str,
+) -> Optional[str]:
+    """
+    Get the _id of a record in DocDb from its name field.
+
+    Parameters
+    ----------
+    client : MetadataDbClient
+    name : str
+
+    Returns
+    -------
+    Optional[str]
+        None if record does not exist. Otherwise, it will return the _id of
+        the record.
+    """
+    records = client.retrieve_docdb_records(
+        filter_query={"name": name}, projection={"_id": 1}, limit=0
+    )
+
+    if len(records) > 1:
+        logging.warning(
+            "Multiple records share the name {name}, ",
+            "only the first record will be returned.",
+        )
+
+    if len(records) > 0:
+        return records[0]["_id"]
+    else:
+        return None