Skip to content

Core Search API Reference

This page documents public, reusable core classes. Collection-specific subclasses are documented in CollectionSearch Subclasses.

NMDC Search Base

The foundational class for cross-collection queries and linked-instance retrieval. Use this class for custom workflows that span multiple schema classes.

NMDCSearch 

NMDCSearch(api_base_url: str = 'https://api.microbiomedata.org', env: str = '')

Bases: nmdc_client.api_client.NMDCAPIClient

Class for interacting with the NMDC Runtime API for searching and retrieving records in the NMDC metadata database.

Parameters:

Name Type Description Default
api_base_url str

The base URL of an instance of the NMDC Runtime API. By default, this is the base URL of the production instance. NMDC team members will occasionally set this to the base URL of a different instance; for example, a self-hosted instance used for testing.

 'https://api.microbiomedata.org'
Warnings

The env parameter is deprecated. Use api_base_url instead.

Source code in nmdc_client/nmdc_search.py 
28
29
def __init__(self, api_base_url: str = API_BASE_URL, env: str = ""):
    super().__init__(api_base_url=api_base_url, env=env)

get_collection_name_from_id 

get_collection_name_from_id(doc_id: str) -> str

Used when you have an id but not the collection name. Determine the collection the id is stored in.

Parameters:

Name Type Description Default
doc_id str

The id of the document.

 required

Returns:

Type Description
str

The collection name of the document.

Raises:

Type Description
RuntimeError

If the API request fails.
Source code in nmdc_client/nmdc_search.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def get_collection_name_from_id(self, doc_id: str) -> str:
    """
    Used when you have an id but not the collection name.
    Determine the collection the id is stored in.

    Parameters
    ----------
    doc_id
        The id of the document.

    Returns
    -------
    str
        The collection name of the document.

    Raises
    ------
    RuntimeError
        If the API request fails.

    """
    url = f"{self.api_base_url}/nmdcschema/ids/{doc_id}/collection-name"
    try:
        response = requests.get(url, headers=self._build_http_request_headers())
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error("API request failed", exc_info=True)
        raise RuntimeError("Failed to get record name from NMDC API") from e
    else:
        logging.debug(
            f"API request response: {response.json()}\n API Status Code: {response.status_code}"
        )

    collection_name = response.json()["collection_name"]
    return collection_name

get_linked_instances 

get_linked_instances(ids: list[str] | str, hydrate: bool = False, types: list[str] | str | None = None, max_page_size: int = 500) -> list[dict]

Retrieve linked instances for the given IDs from the NMDC API.

This method returns a list of linked instance records for the given IDs. For instance, if you provide a study ID, this returns records from the biosample_set, data_generation_set, etc. that are associated with that study, even if the association is not represented by a single direct link.

See get_linked_instances_and_associate_ids for a method that returns an alternate format of the data.

Parameters:

Name Type Description Default
ids list[str] | str

The ids to search for.

 required
hydrate bool

Whether to include full documents in the response.

 False
types list[str] | str | None

The types of records to return. If omitted or None, linked instances of all types are returned. Example: ["nmdc:Study", "nmdc:Biosample", "nmdc:MassSpectrometry"].

 None
max_page_size int

The maximum number of records to return per page.

 500

Returns:

Type Description
list[dict]

A list of linked instance records.
Source code in nmdc_client/nmdc_search.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get_linked_instances(
    self,
    ids: list[str] | str,
    hydrate: bool = False,
    types: list[str] | str | None = None,
    max_page_size: int = 500,
) -> list[dict]:
    """
    Retrieve linked instances for the given IDs from the NMDC API.

    This method returns a list of linked instance records for the given IDs. For instance,
    if you provide a study ID, this returns records from the ``biosample_set``,
    ``data_generation_set``, etc. that are associated with that study, even if the association
    is not represented by a single direct link.

    See ``get_linked_instances_and_associate_ids`` for a method that returns an alternate format of the data.

    Parameters
    ----------
    ids
        The ids to search for.
    hydrate
        Whether to include full documents in the response.
    types
        The types of records to return. If omitted or ``None``, linked instances of all
        types are returned. Example: ["nmdc:Study", "nmdc:Biosample", "nmdc:MassSpectrometry"].
    max_page_size
        The maximum number of records to return per page.

    Returns
    -------
    list[dict]
        A list of linked instance records.
    """
    # highest number I could get to without a timeout
    batch_size = 250
    # Note: We normalize the `ids` value into a list, since the docstring says the caller can
    #       pass it in as either a bare string _or_ a list of strings. If we didn't do this,
    #       and the caller did pass in a bare string, the code below would iterate over the
    #       individual characters of that string (strings are iterable in Python).
    list_of_ids = NMDCSearch._normalize_ids(ids)
    batch_records: list[dict[str, Any]] = []
    url = f"{self.api_base_url}/nmdcschema/linked_instances"
    # split the ids into batches
    for i in range(0, len(list_of_ids), batch_size):
        batch = list_of_ids[i : i + batch_size]
        params = {
            "types": types,
            "ids": batch,
            "hydrate": hydrate,
            "max_page_size": max_page_size,
        }
        response = requests.get(
            url=url,
            params=params,
            headers=self._build_http_request_headers(),
        )
        if response.status_code == 200:
            batch_resources = response.json().get("resources", [])
            next_page = response.json().get("next_page_token", None)
            batch_records.extend(batch_resources)
            if next_page:
                while next_page:
                    params = {
                        "types": types,
                        "ids": batch,
                        "page_token": next_page,
                    }
                    response = requests.get(
                        url=url,
                        params=params,
                        headers=self._build_http_request_headers(),
                    )
                    if response.status_code == 200:
                        batch_resources = response.json().get("resources", [])
                        batch_records.extend(batch_resources)
                        next_page = response.json().get("next_page_token", None)
        else:
            raise RuntimeError(
                f"Error fetching linked instances: {response.status_code} {response.text}"
            )
    return batch_records

get_linked_instances_and_associate_ids 

get_linked_instances_and_associate_ids(ids: list[str] | str, types: list[str] | str | None = None, hydrate: bool = False, max_page_size: int = 500) -> dict[str, list[dict | str]]

Retrieve linked instances for the given IDs from the NMDC API and associate them with the input IDs.

This method returns a list of records that are linked to the records with the given IDs. For instance, if you provide an ID for a study record, this can return the ids records within the biosample_set, data_generation_set etc that are associated with this study, even if it is not a single link between records.

See also get_linked_instances for a method that returns the linked instances in their original list format. This method reformats into a dictionary with keys as query ids, and either a list of resulting linked ids or a list of hydrated records as values.

Parameters:

Name Type Description Default
ids list[str] | str

The ids to search for.

 required
types list[str] | str | None

The types of instances you want to return. If types is None, all types are returned.

 None
hydrate bool

Whether to include full documents in the response.

 False
max_page_size int

The maximum number of records to return per page.

 500

Returns:

Type Description
dict[str, list[dict] | list[str]]

A dictionary mapping each input id to a list of its linked instance records.
Source code in nmdc_client/nmdc_search.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_linked_instances_and_associate_ids(
    self,
    ids: list[str] | str,
    types: list[str] | str | None = None,
    hydrate: bool = False,
    max_page_size: int = 500,
) -> dict[str, list[dict | str]]:
    """
    Retrieve linked instances for the given IDs from the NMDC API and associate them with the input IDs.

    This method returns a list of records that are linked to the records with the given IDs. For instance,
    if you provide an ID for a study record, this can return the ids records within the ``biosample_set``,
    ``data_generation_set`` etc that are associated with this study, even if it is not a single link between records.

    See also ``get_linked_instances`` for a method that returns the linked instances in their original list format.
    This method reformats into a dictionary with keys as query ids, and either a list of resulting linked ids or a list of hydrated records as values.


    Parameters
    ----------
    ids
        The ids to search for.
    types
        The types of instances you want to return. If ``types`` is None, all types are returned.
    hydrate
        Whether to include full documents in the response.
    max_page_size
        The maximum number of records to return per page.

    Returns
    -------
    dict[str, list[dict] | list[str]]
        A dictionary mapping each input id to a list of its linked instance records.
    """
    # get the linked instances
    linked_instances = self.get_linked_instances(
        types=types, ids=ids, hydrate=hydrate, max_page_size=max_page_size
    )
    association: dict[str, list[dict | str]] = {}
    # loop through the linked instances and build the association
    for record in linked_instances:
        for stream in ["_upstream_of", "_downstream_of"]:
            if stream in record:
                for stream_id in record[stream]:
                    if stream_id not in association:
                        association[stream_id] = []
                    if hydrate:
                        association[stream_id].append(
                            {key: record[key] for key in record if key != stream}
                        )
                    else:
                        association[stream_id].append(record["id"])
            else:
                continue

    return association

get_record_from_id 

get_record_from_id(id: str, filter: str = '', fields: str = '') -> dict

Retrieve a record via the NMDC API from a provided record ID.

Parameters:

Name Type Description Default
id str

The ID of the record to retrieve.

 required
filter str

Additional filter to apply to the records. If empty, no additional filter is applied.

 ''
fields str

Comma-separated list of fields to include in the response. If empty, all fields are returned.

 ''

Returns:

Type Description
dict

The full record data.
Source code in nmdc_client/nmdc_search.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def get_record_from_id(self, id: str, filter: str = "", fields: str = "") -> dict:
    """
    Retrieve a record via the NMDC API from a provided record ID.

    Parameters
    ----------
    id
        The ID of the record to retrieve.
    filter
        Additional filter to apply to the records. If empty, no additional filter is applied.
    fields
        Comma-separated list of fields to include in the response. If empty, all fields are returned.

    Returns
    -------
    dict
        The full record data.
    """
    collection_name = self.get_collection_name_from_id(id)
    url = f"{self.api_base_url}/nmdcschema/{collection_name}/{id}"
    params = {
        "filter": filter,
        "projection": fields,
    }
    try:
        response = requests.get(
            url,
            params=params,
            headers=self._build_http_request_headers(),
        )
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error("API request failed", exc_info=True)
        raise RuntimeError(f"Failed to get record {id} from NMDC API") from e
    return response.json()

get_records_by_id 

get_records_by_id(ids: list[str] | str, fields: str = '') -> list[dict]

Retrieve records via the NMDC API from a provided list of record IDs.

The input ids can be from multiple collections. Input like ["nmdc:sty-11-8fb6t785", "nmdc:bsm-11-002vgm56", "nmdc:dobj-11-00095294"] is valid and will return each of these records in a list of dictionaries.

Parameters:

Name Type Description Default
ids list[str] | str

List of IDs of records to retrieve.

 required
fields str

Comma-separated list of fields to include in the response. An empty string returns all fields.

 ''

Returns:

Type Description
list[dict]

The record(s) data.
Source code in nmdc_client/nmdc_search.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def get_records_by_id(
    self,
    ids: list[str] | str,
    fields: str = "",
) -> list[dict]:
    """
    Retrieve records via the NMDC API from a provided list of record IDs.

    The input ids can be from multiple collections. Input like
    ["nmdc:sty-11-8fb6t785", "nmdc:bsm-11-002vgm56", "nmdc:dobj-11-00095294"] is valid and will return each of these records in a list of dictionaries.

    Parameters
    ----------
    ids
        List of IDs of records to retrieve.
    fields
        Comma-separated list of fields to include in the response. An empty string returns all fields.

    Returns
    -------
    list[dict]
        The record(s) data.
    """

    resources: list[dict[str, Any]] = []
    # sort the input ids
    sorted_ids = sorted(ids) if isinstance(ids, list) else [ids]
    id_dict: dict[str, list[str]] = {}
    # group ids by their collection subset nmdc:sty, nmdc:bsm, etc
    for id in sorted_ids:
        cur_group = id.split("-")[0]
        if cur_group not in id_dict:
            id_dict[cur_group] = []
        id_dict[cur_group].append(id)

    for cur_group in id_dict:
        # process each group of ids
        id_list = id_dict[cur_group]
        # for each group, get the collection name from one of the ids
        collection_name = self.get_collection_name_from_id(id_list[0])
        # import in function to circumvent circular import error
        from nmdc_client.collection_search import CollectionSearch

        cs = CollectionSearch(
            collection_name=collection_name, api_base_url=self.api_base_url
        )
        records = cs.get_batch_records(
            id_list=id_list,
            search_field="id",
            fields=fields,
        )
        resources.extend(records)
    return resources

get_schema_version 

get_schema_version() -> str

Get the current NMDC schema version used by the NMDC API.

Returns:

Type Description
str

The NMDC schema version
Source code in nmdc_client/nmdc_search.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def get_schema_version(self) -> str:
    """
    Get the current NMDC schema version used by the NMDC API.

    Returns
    -------
    str
        The NMDC schema version
    """

    url = f"{self.api_base_url}/version"
    try:
        response = requests.get(url, headers=self._build_http_request_headers())
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error("API request failed", exc_info=True)
        raise RuntimeError("Failed to version from NMDC API") from e
    return response.json()["nmdc-schema"]

Collection Search Base

Extends NMDCSearch with collection-focused query helpers. Use this class for generic collection operations, or use CollectionSearch Subclasses for preconfigured collection targets.

CollectionSearch 

CollectionSearch(collection_name: str, api_base_url: str = 'https://api.microbiomedata.org', env: str = '')

Bases: nmdc_client.nmdc_search.NMDCSearch

Class to interact with the NMDC API to search for records within a specified collection.

Parameters:

Name Type Description Default
collection_name str

The name of the collection to search within.

 required
api_base_url str

The base URL of an instance of the NMDC Runtime API. By default, this is the base URL of the production instance.

 'https://api.microbiomedata.org'
Warnings

The env parameter is deprecated. Use api_base_url instead.

Source code in nmdc_client/collection_search.py 
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    collection_name: str,
    api_base_url: str = API_BASE_URL,
    env: str = "",
):
    self.collection_name = collection_name
    super().__init__(
        api_base_url=api_base_url,
        env=env,
    )

build_filter 

build_filter(attributes: dict[str, str], exact_match: bool = False) -> str

Build a MongoDB-style query filter string from one or more attributes.

The resulting string will be compatible with the filter parameter of various NMDC API endpoints.

Parameters:

Name Type Description Default
attributes dict[str, str]

Dictionary of attribute names and their corresponding values.

 required
exact_match bool

Whether attribute values should be exact (i.e. case sensitive) or inexact (i.e. case insensitive) matches.

 False

Returns:

Type Description
str

A string representing the MongoDB filter.
Source code in nmdc_client/collection_search.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def build_filter(
    self, attributes: dict[str, str], exact_match: bool = False
) -> str:
    """
    Build a MongoDB-style query filter string from one or more attributes.

    The resulting string will be compatible with the `filter` parameter
    of various NMDC API endpoints.

    Parameters
    ----------
    attributes
        Dictionary of attribute names and their corresponding values.
    exact_match
        Whether attribute values should be exact (i.e. case sensitive)
        or inexact (i.e. case insensitive) matches.

    Returns
    -------
    str
        A string representing the MongoDB filter.
    """
    filter_dict: dict[str, str | dict[str, str]] = {}
    if exact_match:
        for attribute_name, attribute_value in attributes.items():
            filter_dict[attribute_name] = attribute_value
    else:
        for attribute_name, attribute_value in attributes.items():
            # escape special characters - mongo db filters require special characters to be double escaped ex. GC\\-MS \\(2009\\)
            escaped_value = re.sub(r"([\W])", r"\\\1", attribute_value)
            filter_dict[attribute_name] = {"$regex": escaped_value, "$options": "i"}

    return json.dumps(filter_dict, separators=(",", ":"))

check_ids_exist 

check_ids_exist(ids: list[str], chunk_size: int = 100, return_missing_ids: bool = False) -> bool | tuple[bool, list[str]]

Check if specified IDs exist in the collection.

This method constructs a query to the API to filter the collection based on the given IDs, and checks if all IDs exist in the collection.

Parameters:

Name Type Description Default
ids list[str]

A list of IDs to check if they exist in the collection.

 required
chunk_size int

The number of IDs to check in each query.

 100
return_missing_ids bool

If True, and if ids are missing in the collection, return the list of IDs that do not exist in the collection.

 False

Returns:

Type Description
bool | tuple[bool, list[str]]

True if all IDs exist in the collection, False otherwise. However, if return_missing_ids is True, returns a tuple whose first item is the aforementioned boolean value, and whose second item is a list of the IDs, if any, that don't exist in the collection.
Source code in nmdc_client/collection_search.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def check_ids_exist(
    self,
    ids: list[str],
    chunk_size: int = 100,
    return_missing_ids: bool = False,
) -> bool | tuple[bool, list[str]]:
    """
    Check if specified IDs exist in the collection.

    This method constructs a query to the API to filter the collection based on the given IDs, and checks if all IDs exist in the collection.

    Parameters
    ----------
    ids
        A list of IDs to check if they exist in the collection.
    chunk_size
        The number of IDs to check in each query.
    return_missing_ids
        If True, and if ids are missing in the collection, return the list of IDs that do not exist in the collection.

    Returns
    -------
    bool | tuple[bool, list[str]]
        True if all IDs exist in the collection, False otherwise. However,
        if return_missing_ids is True, returns a tuple whose first item is the aforementioned boolean value,
        and whose second item is a list of the IDs, if any, that don't exist in the collection.
    """
    if not getattr(self, "supports_get_by_id", True):
        raise OperationNotSupportedError(
            f"check_ids_exist is not supported for the {self.collection_name} collection"
        )

    # chunk the input list of IDs into smaller lists (each of size `chunk_size`)
    # to avoid the maximum URL length limit
    ids_test = list(set(ids))
    for i in range(0, len(ids_test), chunk_size):
        chunk = ids_test[i : i + chunk_size]
        filter_dict = {"id": {"$in": chunk}}
        filter_json_string = json.dumps(filter_dict, separators=(",", ":"))

        results = self.get_records(
            filter=filter_json_string,
            max_page_size=len(chunk),
            fields="id",
        )
        if len(results) != len(chunk) and return_missing_ids:
            missing_ids = list(
                set(chunk) - set([record["id"] for record in results])
            )
            return False, missing_ids
        elif len(results) != len(chunk) and not return_missing_ids:
            return False
    return True

get_batch_records 

get_batch_records(id_list: list, search_field: str, chunk_size: int = 100, fields: str = '') -> list[dict]

Get a batch of records from the collection that relate to input IDs.

This method is used to retrieve records that include any of the IDs from the input list in specified fields (including fields other than id). For example, if records in a collection contain study IDs in a field called associated_studies, this method can be used to retrieve all records that include any of the input study IDs in the associated_studies field.

Parameters:

Name Type Description Default
id_list list

A list of IDs to get records for.

 required
search_field str

The field in which to search for the IDs.

 required
chunk_size int

The number of IDs to get in each query.

 100
fields str

The fields to return. If empty or not provided, all fields are returned.

 ''

Returns:

Type Description
list[dict]

A list of dictionaries (still packed) containing the records that relate to the input IDs in the specified search field.
Source code in nmdc_client/collection_search.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def get_batch_records(
    self,
    id_list: list,
    search_field: str,
    chunk_size: int = 100,
    fields: str = "",
) -> list[dict]:
    """
    Get a batch of records from the collection that relate to input IDs.

    This method is used to retrieve records that include any of the IDs from the input list in specified fields (including fields other than ``id``).
    For example, if records in a collection contain study IDs in a field called ``associated_studies``,
    this method can be used to retrieve all records that include any of the input study IDs in the ``associated_studies`` field.

    Parameters
    ---------
    id_list
        A list of IDs to get records for.
    search_field
        The field in which to search for the IDs.
    chunk_size
        The number of IDs to get in each query.
    fields
        The fields to return. If empty or not provided, all fields are returned.
    Returns
    -------
    list[dict]
        A list of dictionaries (still packed) containing the records that relate to the input IDs in
        the specified search field.
    """
    if not getattr(self, "supports_get_by_id", True):
        raise OperationNotSupportedError(
            f"get_record_by_id is not supported for the {self.collection_name} collection"
        )

    results: list[dict] = []
    id_list = list(set(id_list))
    for i in range(0, len(id_list), chunk_size):
        chunk = id_list[i : i + chunk_size]
        sanitized_chunk = json.dumps(chunk)
        filter = f'{{"{search_field}": {{"$in": {sanitized_chunk}}}}}'
        res = self.get_records(
            filter=filter, max_page_size=len(chunk), fields=fields, all_pages=True
        )
        results += res
    return results

get_record_by_attribute 

get_record_by_attribute(attribute_name: str, attribute_value: str, max_page_size: int = 25, fields: str = '', all_pages: bool = True, exact_match: bool = False) -> list[dict]

Retrieve a record via the NMDC API by a specific attribute's value.

Parameters:

Name Type Description Default
attribute_name str

The name of the attribute to filter by.

 required
attribute_value str

The value of the attribute to filter by.

 required
max_page_size int

The number of records to return per page.

 25
fields str

The fields to return. If empty, all fields are returned.

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True
exact_match bool

Whether the attribute value should be matched exactly or partially. Used to determine if the inputted attribute value is an exact match or a partial match. Default is False, meaning the user does not need to input an exact match.

 False

Returns:

Type Description
list[dict]

A list of dictionaries containing the records.
Source code in nmdc_client/collection_search.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def get_record_by_attribute(
    self,
    attribute_name: str,
    attribute_value: str,
    max_page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
    exact_match: bool = False,
) -> list[dict]:
    """
    Retrieve a record via the NMDC API by a specific attribute's value.

    Parameters
    ----------
    attribute_name
        The name of the attribute to filter by.
    attribute_value
        The value of the attribute to filter by.
    max_page_size
        The number of records to return per page.
    fields
        The fields to return. If empty, all fields are returned.
    all_pages
        True to return all pages. False to return only the first page.
    exact_match
        Whether the attribute value should be matched exactly or partially.
        Used to determine if the inputted attribute value is an exact match or a partial match.
        Default is False, meaning the user does not need to input an exact match.
    Returns
    -------
    list[dict]
        A list of dictionaries containing the records.

    """
    filter = self.build_filter(
        attributes={attribute_name: attribute_value},
        exact_match=exact_match,
    )

    logging.debug(f"get_record_by_attribute Filter: {filter}")
    results = self.get_records(filter, max_page_size, fields, all_pages)
    return results

get_record_by_filter 

get_record_by_filter(filter: str, max_page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve a record via the NMDC API using a specified filter.

Parameters:

Name Type Description Default
filter str

The filter to use to query the collection. Must be in MongoDB query format. Example: {"name":"my record name"}. More resources for constructing MongoDB filters can be found here <https://www.mongodb.com/docs/manual/reference/method/db.collection.find/#std-label-method-find-query>_.

 required
max_page_size int

The number of records to return per page.

 25
fields str

The fields to return. Default will return all fields. Example: "id,name,description,url,type"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of dictionaries containing the records.
Source code in nmdc_client/collection_search.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def get_record_by_filter(
    self,
    filter: str,
    max_page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve a record via the NMDC API using a specified filter.

    Parameters
    ----------
    filter
        The filter to use to query the collection. Must be in MongoDB query format.
        Example: {"name":"my record name"}.
        `More resources for constructing MongoDB filters can be found here <https://www.mongodb.com/docs/manual/reference/method/db.collection.find/#std-label-method-find-query>`_.
    max_page_size
        The number of records to return per page.
    fields
        The fields to return. Default will return all fields.
        Example: "id,name,description,url,type"
    all_pages
        True to return all pages. False to return only the first page.
    Returns
    -------
    list[dict]
        A list of dictionaries containing the records.

    """
    results = self.get_records(filter, max_page_size, fields, all_pages)
    return results

get_record_by_id 

get_record_by_id(record_id: Optional[str] = None, max_page_size: int = 100, fields: str = '', collection_id: Optional[str] = None) -> list[dict]

Retrieve a record from the collection via the NMDC API using a specified ID.

Parameters:

Name Type Description Default
record_id Optional[str]

The id of the record to retrieve from the collection. Not required to enable backwards compatibility with the deprecated collection_id parameter.

 None
max_page_size int

The maximum number of records to return per page. Default is 100.

 100
fields str

The fields to return. Default is all fields.

 ''
collection_id Optional[str]

The id of the record to retrieve from the collection. This parameter is deprecated and will be removed in a future version. Please use record_id instead.

 None

Returns:

Type Description
list[dict]

A list containing the record. The API typically returns a single dictionary for this endpoint, and this method normalizes that response into a single-item list.

Raises:

Type Description
RuntimeError

If the API request fails.
Source code in nmdc_client/collection_search.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
@has_deprecated_parameter("collection_id", reason="Use ``record_id`` instead.")
def get_record_by_id(
    self,
    record_id: Optional[str] = None,
    max_page_size: int = 100,
    fields: str = "",
    collection_id: Optional[str] = None,
) -> list[dict]:
    """
    Retrieve a record from the collection via the NMDC API using a specified ID.

    Parameters
    ----------
    record_id:
        The id of the record to retrieve from the collection. Not required to enable backwards compatibility with the deprecated collection_id parameter.
    max_page_size:
        The maximum number of records to return per page. Default is 100.
    fields:
        The fields to return. Default is all fields.
    collection_id:
        The id of the record to retrieve from the collection. This parameter is deprecated and will be removed in a future version. Please use record_id instead.
    Returns
    -------
    list[dict]
        A list containing the record. The API typically returns a single dictionary for this
        endpoint, and this method normalizes that response into a single-item list.

    Raises
    ------
    RuntimeError
        If the API request fails.

    """
    if not getattr(self, "supports_get_by_id", True):
        raise OperationNotSupportedError(
            f"get_record_by_id is not supported for the {self.collection_name} collection"
        )

    if record_id is None and collection_id is None:
        raise ValueError(
            "No record_id provided. Please provide this parameter to retrieve a record."
        )
    if record_id and collection_id:
        raise ValueError(
            "Both record_id and collection_id were provided. Please provide record_id, as collection_id is deprecated and will be removed in a future version."
        )
    if collection_id:
        record_id = collection_id

    url = f"{self.api_base_url}/nmdcschema/{self.collection_name}/{record_id}"
    params: dict[str, QueryParamValue] = {
        "max_page_size": max_page_size,
        "projection": fields,
    }
    # get the reponse
    try:
        response = requests.get(
            url=url,
            headers=self._build_http_request_headers(),
            params=params,
        )
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error("API request failed", exc_info=True)
        raise RuntimeError("Failed to get collection by id from NMDC API") from e
    else:
        logging.debug(
            f"API request response: {response.json()}\n API Status Code: {response.status_code}"
        )
    results = response.json()
    if isinstance(results, dict):
        return [results]
    return results

get_records 

get_records(filter: str = '', max_page_size: int = 100, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records from the collection via the NMDC API.

Parameters:

Name Type Description Default
filter str

The filter to apply to the query. An empty string will return all records.

 ''
max_page_size int

The maximum number of records to return per page.

 100
fields str

The fields to return. An empty string will return all fields.

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of dictionaries containing the records.

Raises:

Type Description
RuntimeError

If the API request fails.
Source code in nmdc_client/collection_search.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_records(
    self,
    filter: str = "",
    max_page_size: int = 100,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records from the collection via the NMDC API.

    Parameters
    ----------
    filter
        The filter to apply to the query. An empty string will return all records.
    max_page_size
        The maximum number of records to return per page.
    fields
        The fields to return. An empty string will return all fields.
    all_pages
        True to return all pages. False to return only the first page.
    Returns
    -------
    list[dict]
        A list of dictionaries containing the records.

    Raises
    ------
    RuntimeError
        If the API request fails.

    """
    url = f"{self.api_base_url}/nmdcschema/{self.collection_name}"
    params: dict[str, QueryParamValue] = {
        "filter": filter,
        "max_page_size": max_page_size,
        "projection": fields,
    }
    try:
        response = requests.get(
            url=url,
            params=params,
            headers=self._build_http_request_headers(),
        )
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error("API request failed", exc_info=True)
        raise RuntimeError("Failed to get collection from NMDC API") from e
    else:
        logging.debug(
            f"API request response: {response.json()}\n API Status Code: {response.status_code}"
        )

    results = response.json()["resources"]
    # otherwise, get all pages
    if all_pages:
        results = self._get_all_pages(response, url, filter, max_page_size, fields)[
            "resources"
        ]

    return results

Provides search utilities focused on functional annotation and related retrieval patterns.

FunctionalSearch 

FunctionalSearch(api_base_url: str = 'https://api.microbiomedata.org', env: str = '')

Bases: nmdc_client.collection_search.CollectionSearch

Class to interact with the NMDC API to search for records within the functional_annotation_agg collection.

Warnings

The env parameter is deprecated. Use api_base_url instead.

Source code in nmdc_client/functional_search.py 
16
17
18
19
20
21
def __init__(self, api_base_url: str = API_BASE_URL, env: str = ""):
    super().__init__(
        collection_name="functional_annotation_agg",
        api_base_url=api_base_url,
        env=env,
    )

supports_get_by_id class-attribute 

supports_get_by_id = False

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

get_functional_annotations 

get_functional_annotations(annotation: str, annotation_type: str, page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records with specific annotation value and type.

Parameters:

Name Type Description Default
annotation str

The functional annotation value to query.

 required
annotation_type str

The type of id to query. See Notes for more details.

 required
page_size int

The number of results to return per page.

 25
fields str

The fields to return. If empty, all fields are returned. Example: "id,name"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of functional annotations.

Raises:

Type Description
ValueError

If the annotation_type is not one of the allowed types. See Notes for more details.
Notes

The annotation_type must be one of the following: "KEGG", "COG", "PFAM".

Source code in nmdc_client/functional_search.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_functional_annotations(
    self,
    annotation: str,
    annotation_type: str,
    page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records with specific annotation value and type.

    Parameters
    -----------
    annotation
        The functional annotation value to query.
    annotation_type
        The type of id to query. See Notes for more details.
    page_size
        The number of results to return per page.
    fields
        The fields to return. If empty, all fields are returned.
        Example: "id,name"
    all_pages
        True to return all pages. False to return only the first page.

    Returns
    -------
    list[dict]
        A list of functional annotations.

    Raises
    ------
    ValueError
        If the annotation_type is not one of the allowed types. See Notes for more details.

    Notes
    -----
    The ``annotation_type`` must be one of the following: "KEGG", "COG", "PFAM".
    """
    if annotation_type not in ["KEGG", "COG", "PFAM"]:
        raise ValueError(
            "annotation_type must be one of the following: KEGG, COG, PFAM"
        )
    if annotation_type == "KEGG":
        formatted_annotation_type = f"KEGG.ORTHOLOGY:{annotation}"
    elif annotation_type == "COG":
        formatted_annotation_type = f"COG:{annotation}"
    elif annotation_type == "PFAM":
        formatted_annotation_type = f"PFAM:{annotation}"

    filter = f'{{"gene_function_id": "{formatted_annotation_type}"}}'

    return self.get_record_by_filter(filter, page_size, fields, all_pages)

Latitude and Longitude Utilities

Provides geospatial helper methods for lat/lon-based filtering and coordinate handling.

LatLongFilters

Bases: abc.ABC

Mixin class with methods to interact with collections that can be searched by latitude and longitude via the NMDC API.

get_record_by_lat_long 

get_record_by_lat_long(lat_comparison: str, long_comparison: str, latitude: float, longitude: float, page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records by latitude and longitude filters via the NMDC API.

Parameters:

Name Type Description Default
lat_comparison str

The comparison to use to query the record for latitude. See Notes for more details.

 required
long_comparison str

The comparison to use to query the record for longitude. See Notes for more details.

 required
latitude float

The latitude of the record to query.

 required
longitude float

The longitude of the record to query.

 required
page_size int

The number of results to return per page.

 25
fields str

The fields to return. If empty, all fields are returned. Example: "id,name,description,type"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of records.

Raises:

Type Description
ValueError

If the comparison is not one of the allowed comparisons.
Notes

lat_comparison and long_comparison must be one of the following:

  • eq : Matches values that are equal to the given value.
  • gt : Matches if values are greater than the given value.
  • lt : Matches if values are less than the given value.
  • gte : Matches if values are greater or equal to the given value.
  • lte : Matches if values are less or equal to the given value.
Source code in nmdc_client/lat_long_filters.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def get_record_by_lat_long(
    self,
    lat_comparison: str,
    long_comparison: str,
    latitude: float,
    longitude: float,
    page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records by latitude and longitude filters via the NMDC API.

    Parameters
    ----------
    lat_comparison
        The comparison to use to query the record for latitude. See Notes for more details.
    long_comparison
        The comparison to use to query the record for longitude. See Notes for more details.
    latitude
        The latitude of the record to query.
    longitude
        The longitude of the record to query.
    page_size
        The number of results to return per page.
    fields
        The fields to return. If empty, all fields are returned.
        Example: "id,name,description,type"
    all_pages
        True to return all pages. False to return only the first page.

    Returns
    -------
    list[dict]
        A list of records.

    Raises
    ------
    ValueError
        If the comparison is not one of the allowed comparisons.

    Notes
    -----
    ``lat_comparison`` and ``long_comparison`` must be one of the following:

    - eq : Matches values that are equal to the given value.
    - gt : Matches if values are greater than the given value.
    - lt : Matches if values are less than the given value.
    - gte : Matches if values are greater or equal to the given value.
    - lte : Matches if values are less or equal to the given value.

    """
    allowed_comparisons = ["eq", "gt", "lt", "gte", "lte"]
    if lat_comparison not in allowed_comparisons:
        logger.error(
            f"Invalid comparison input: {lat_comparison}\n Valid inputs: {allowed_comparisons}"
        )
        raise ValueError(
            f"Invalid comparison input: {lat_comparison}\n Valid inputs: {allowed_comparisons}"
        )
    if long_comparison not in allowed_comparisons:
        logger.error(
            f"Invalid comparison input: {long_comparison}\n Valid inputs: {allowed_comparisons}"
        )
        raise ValueError(
            f"Invalid comparison input: {long_comparison}\n Valid inputs: {allowed_comparisons}"
        )
    filter = f'{{"lat_lon.latitude": {{"${lat_comparison}": {latitude}}}, "lat_lon.longitude": {{"${long_comparison}": {longitude}}}}}'
    results = self.get_records(filter, page_size, fields, all_pages)
    return results

get_record_by_latitude 

get_record_by_latitude(comparison: str, latitude: float, page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records by latitude filter via the NMDC API.

Parameters:

Name Type Description Default
comparison str

The comparison to use to query the record. See Notes for more details.

 required
latitude float

The latitude of the record to query.

 required
page_size int

The number of results to return per page.

 25
fields str

The fields to return. Default is all fields. Example: "id,name,description,type"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of records.

Raises:

Type Description
ValueError

If the comparison is not one of the allowed comparisons.
Notes

The comparison must be one of the following: "eq", "gt", "lt", "gte", "lte".

  • eq : Matches values that are equal to the given value.
  • gt : Matches if values are greater than the given value.
  • lt : Matches if values are less than the given value.
  • gte : Matches if values are greater or equal to the given value.
  • lte : Matches if values are less or equal to the given value.
Source code in nmdc_client/lat_long_filters.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_record_by_latitude(
    self,
    comparison: str,
    latitude: float,
    page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records by latitude filter via the NMDC API.

    Parameters
    ----------
    comparison
        The comparison to use to query the record. See Notes for more details.
    latitude
        The latitude of the record to query.
    page_size
        The number of results to return per page.
    fields
        The fields to return. Default is all fields.
        Example: "id,name,description,type"
    all_pages
        True to return all pages. False to return only the first page.

    Returns
    -------
    list[dict]
        A list of records.

    Raises
    ------
    ValueError
        If the comparison is not one of the allowed comparisons.

    Notes
    -----
    The ``comparison`` must be one of the following: "eq", "gt", "lt", "gte", "lte".

    - eq : Matches values that are equal to the given value.
    - gt : Matches if values are greater than the given value.
    - lt : Matches if values are less than the given value.
    - gte : Matches if values are greater or equal to the given value.
    - lte : Matches if values are less or equal to the given value.

    """
    allowed_comparisons = ["eq", "gt", "lt", "gte", "lte"]
    if comparison not in allowed_comparisons:
        logger.error(
            f"Invalid comparison input: {comparison}\n Valid inputs: {allowed_comparisons}"
        )
        raise ValueError(
            f"Invalid comparison input: {comparison}\n Valid inputs: {allowed_comparisons}"
        )
    filter = f'{{"lat_lon.latitude": {{"${comparison}": {latitude}}}}}'

    result = self.get_records(filter, page_size, fields, all_pages)
    return result

get_record_by_longitude 

get_record_by_longitude(comparison: str, longitude: float, page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records by longitude filter via the NMDC API.

Parameters:

Name Type Description Default
comparison str

The comparison to use to query the record. See Notes for more details.

 required
longitude float

The longitude of the record to query.

 required
page_size int

The number of results to return per page.

 25
fields str

The fields to return. If empty, all fields are returned. Example: "id,name,description,type"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of records.

Raises:

Type Description
ValueError

If the comparison is not one of the allowed comparisons.
Notes

The comparison must be one of the following: "eq", "gt", "lt", "gte", "lte".

  • eq : Matches values that are equal to the given value.
  • gt : Matches if values are greater than the given value.
  • lt : Matches if values are less than the given value.
  • gte : Matches if values are greater or equal to the given value.
  • lte : Matches if values are less or equal to the given value.
Source code in nmdc_client/lat_long_filters.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_record_by_longitude(
    self,
    comparison: str,
    longitude: float,
    page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records by longitude filter via the NMDC API.

    Parameters
    ----------
    comparison
        The comparison to use to query the record. See Notes for more details.
    longitude
        The longitude of the record to query.
    page_size
        The number of results to return per page.
    fields
        The fields to return. If empty, all fields are returned.
        Example: "id,name,description,type"
    all_pages
        True to return all pages. False to return only the first page.

    Returns
    -------
    list[dict]
        A list of records.

    Raises
    ------
    ValueError
        If the comparison is not one of the allowed comparisons.

    Notes
    -----
    The ``comparison`` must be one of the following: "eq", "gt", "lt", "gte", "lte".

    - eq : Matches values that are equal to the given value.
    - gt : Matches if values are greater than the given value.
    - lt : Matches if values are less than the given value.
    - gte : Matches if values are greater or equal to the given value.
    - lte : Matches if values are less or equal to the given value.
    """
    allowed_comparisons = ["eq", "gt", "lt", "gte", "lte"]
    if comparison not in allowed_comparisons:
        logger.error(
            f"Invalid comparison input: {comparison}\n Valid inputs: {allowed_comparisons}"
        )
        raise ValueError(
            f"Invalid comparison input: {comparison}\n Valid inputs: {allowed_comparisons}"
        )
    filter = f'{{"lat_lon.longitude": {{"${comparison}": {longitude}}}}}'
    result = self.get_records(filter, page_size, fields, all_pages)
    return result

get_record_by_proximity 

get_record_by_proximity(radius_km: float, record_id: str | None = None, query_lat: float | None = None, query_lon: float | None = None, page_size: int = 25, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records by proximity to a record or lat lon via the NMDC API. First retrieve records within a bounding box, then refine to those within the circular radius.

Parameters:

Name Type Description Default
radius_km float

The radius in kilometers to search for records around the record.

 required
record_id str | None

The ID of a record to query where lat/lon are provided (Biosample or FieldResearchSite). If provided, query_lat and query_lon must not be provided.

 None
query_lat float | None

The latitude, in decimal degrees, to query. If provided, record_id must not be provided. Example: 63.875088

 None
query_lon float | None

The longitude, in decimal degrees, to query. If provided, record_id must not be provided. Example: -149.210438

 None
page_size int

The number of results to return per page.

 25
fields str

The fields to return. If empty, all fields are returned. Example: "id,name,description,type"

 ''
all_pages bool

True to return all pages. False to return only the first page.

 True

Returns:

Type Description
list[dict]

A list of records.
Source code in nmdc_client/lat_long_filters.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def get_record_by_proximity(
    self,
    radius_km: float,
    record_id: str | None = None,
    query_lat: float | None = None,
    query_lon: float | None = None,
    page_size: int = 25,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """
    Retrieve records by proximity to a record or lat lon via the NMDC API.
    First retrieve records within a bounding box, then refine to those within the circular radius.

    Parameters
    ----------
    radius_km
        The radius in kilometers to search for records around the record.
    record_id
        The ID of a record to query where lat/lon are provided (Biosample or FieldResearchSite). If provided, query_lat and query_lon must not be provided.
    query_lat
        The latitude, in decimal degrees, to query. If provided, record_id must not be provided.
        Example: 63.875088
    query_lon
        The longitude, in decimal degrees, to query. If provided, record_id must not be provided.
        Example: -149.210438
    page_size
        The number of results to return per page.
    fields
        The fields to return. If empty, all fields are returned.
        Example: "id,name,description,type"
    all_pages
        True to return all pages. False to return only the first page.

    Returns
    -------
    list[dict]
        A list of records.

    """
    radius_meters = radius_km * 1000

    if record_id is not None and (query_lat is not None or query_lon is not None):
        logger.error(
            "Invalid input: ONLY record_id or query_lat/query_lon can be used, not both."
        )
        raise ValueError(
            "Invalid input: ONLY record_id or query_lat/query_lon can be used, not both."
        )
    if (
        query_lat is not None
        and query_lon is None
        or query_lat is None
        and query_lon is not None
    ):
        logger.error(
            "Invalid input: BOTH query_lat and query_lon must be provided."
        )
        raise ValueError(
            "Invalid input: BOTH query_lat and query_lon must be provided."
        )

    # Calculate bounding box for mongoDB query
    if record_id is not None:
        record_lat_lon = self.get_records(
            filter=f'{{"id": "{record_id}"}}',
            max_page_size=1,
            fields="lat_lon",
            all_pages=False,
        )
        center_lat = record_lat_lon[0].get("lat_lon", {}).get("latitude")
        center_lon = record_lat_lon[0].get("lat_lon", {}).get("longitude")
    if query_lat is not None and query_lon is not None:
        center_lat = query_lat
        center_lon = query_lon
    min_lat, max_lat, min_lon, max_lon = self._bounding_box(
        center_lat, center_lon, radius_meters
    )

    # MongoDB query with bounding box
    filter = f'{{"lat_lon.latitude": {{"$gte": {min_lat}, "$lte": {max_lat}}},"lat_lon.longitude": {{"$gte": {min_lon}, "$lte": {max_lon}}}}}'
    results = cast(
        list[dict],
        self.get_records(filter, page_size, fields, all_pages),
    )

    # Refine results to those within circular radius
    refined_results = []
    for record in results:
        lat_lon = record.get("lat_lon")
        if not isinstance(lat_lon, dict):
            continue
        rec_lat = lat_lon.get("latitude")
        rec_lon = lat_lon.get("longitude")
        if rec_lat is not None and rec_lon is not None:
            distance = self._haversine_distance_m(
                center_lat, center_lon, rec_lat, rec_lon
            )
            if distance <= radius_meters:
                refined_results.append(record)

    return cast(list[dict], refined_results)

get_records 

get_records(filter: str = '', max_page_size: int = 100, fields: str = '', all_pages: bool = True) -> list[dict]

Retrieve records from a collection via the NMDC API.

Source code in nmdc_client/lat_long_filters.py 
16
17
18
19
20
21
22
23
24
@abstractmethod
def get_records(
    self,
    filter: str = "",
    max_page_size: int = 100,
    fields: str = "",
    all_pages: bool = True,
) -> list[dict]:
    """Retrieve records from a collection via the NMDC API."""

General Utilities

General-purpose helper utilities used across workflows.

Utils 

Utils()
Source code in nmdc_client/utils.py 
8
9
def __init__(self):
    pass