nexusLIMS.harvesters.nemo.utils#

Various utility functions used by the NEMO harvester.

Module Contents#

Functions#

get_harvesters_enabled

Return a list of enabled connectors based off the environment.

add_all_usage_events_to_db

Add all usage events to database for enabled NEMO connectors.

get_usage_events_as_sessions

Get all usage events for enabled NEMO connectors as Sessions.

get_connector_for_session

Get the appropriate NEMO connector for a given Session.

get_connector_by_base_url

Get an enabled NemoConnector by inspecting the base_url.

process_res_question_samples

Process sample information from reservation questions.

has_valid_question_data

Check if usage event has valid question data in specified field.

id_from_url

Get the value of the id query parameter stored in URL string.

API#

nexusLIMS.harvesters.nemo.utils.get_harvesters_enabled() → List[NemoConnector][source]#

Return a list of enabled connectors based off the environment.

Returns:

harvesters_enabled – A list of NemoConnector objects representing the NEMO APIs enabled via environment settings

Return type:

List[NemoConnector]

nexusLIMS.harvesters.nemo.utils.add_all_usage_events_to_db(user: Union[str, int] | None = None, dt_from: datetime | None = None, dt_to: datetime | None = None, tool_id: Union[int, List[int]] | None = None)[source]#

Add all usage events to database for enabled NEMO connectors.

Loop through enabled NEMO connectors and add each one’s usage events to the NexusLIMS session_log database table (if required).

Parameters:

  • user – The user(s) for which to add usage events. If None, events will not be filtered by user at all

  • dt_from – The point in time after which usage events will be added. If None, no date filtering will be performed

  • dt_to – The point in time before which usage events will be added. If None, no date filtering will be performed

  • tool_id – The tools(s) for which to add usage events. If 'None' (default), the tool IDs for each instrument in the NexusLIMS DB will be extracted and used to limit the API response

nexusLIMS.harvesters.nemo.utils.get_usage_events_as_sessions(user: Union[str, int] | None = None, dt_from: datetime | None = None, dt_to: datetime | None = None, tool_id: Union[int, List[int]] | None = None) → List[Session][source]#

Get all usage events for enabled NEMO connectors as Sessions.

Loop through enabled NEMO connectors and return each one’s usage events to as Session objects without writing logs to the session_log table. Mostly used for doing dry runs of the record builder.

Parameters:

  • user – The user(s) for which to fetch usage events. If None, events will not be filtered by user at all

  • dt_from – The point in time after which usage events will be fetched. If None, no date filtering will be performed

  • dt_to – The point in time before which usage events will be fetched. If None, no date filtering will be performed

  • tool_id – The tools(s) for which to fetch usage events. If None, events will only be filtered by tools known in the NexusLIMS DB for each connector

nexusLIMS.harvesters.nemo.utils.get_connector_for_session(session: Session) → NemoConnector[source]#

Get the appropriate NEMO connector for a given Session.

Given a Session, find the matching NemoConnector from the enabled list of NEMO harvesters.

Parameters:

session – The session for which a NemoConnector is needed

Returns:

n – The connector object that allows for querying the NEMO API for the instrument contained in session

Return type:

NemoConnector

Raises:

LookupError – Raised if a matching connector is not found

nexusLIMS.harvesters.nemo.utils.get_connector_by_base_url(base_url: str) → NemoConnector[source]#

Get an enabled NemoConnector by inspecting the base_url.

Parameters:

base_url – A portion of the API url to search for

Returns:

n – The enabled NemoConnector instance

Return type:

NemoConnector

Raises:

LookupError – Raised if a matching connector is not found

nexusLIMS.harvesters.nemo.utils.process_res_question_samples(res_dict: Dict) → Tuple[List[str | None] | None, List[str | None] | None, List[str | None] | None, List[str | None] | None][source]#

Process sample information from reservation questions.

Parameters:

res_dict – The reservation dictionary (i.e. the response from the reservations api endpoint)

nexusLIMS.harvesters.nemo.utils.has_valid_question_data(event_dict: Dict, field: str = 'run_data') → bool[source]#

Check if usage event has valid question data in specified field.

Works for both run_data and pre_run_data fields, which use identical structure.

A usage event has valid question data if:

  1. It has the specified field

  2. Field value is not None or empty string

  3. Field value can be parsed as JSON

  4. Parsed data is not empty

  5. Parsed data has data_consent field (required)

Parameters:

  • event_dict – The usage event dictionary from NEMO API

  • field – Field name to check (“run_data” or “pre_run_data”)

Returns:

True if usage event has valid question data in the specified field, False otherwise

Return type:

bool

nexusLIMS.harvesters.nemo.utils.id_from_url(url: str) → int | None[source]#

Get the value of the id query parameter stored in URL string.

This is used to extract the value as needed from API strings.

Parameters:

url – The URL to parse, such as https://nemo.example.com/api/usage_events/?id=9

Returns:

this_id – The id value if one is present, otherwise None

Return type:

None or int