mgtx-benchling-wrapper

A wrapper of the Benchling API SDK with common functions and workflows used at MGTX DSC.

Installation 🚀

You can install this package using:

pip install mgtx-benchling-wrapper

Quickstart 🚩

After creating an app on your Benchling tenant, create a config.yaml file in your repo. The following is an example of the contents of a config.yaml

Note: You might want to keep the app client secret separated from your main code.

BenchlingCredentials:
  benchling_url: 'https://mytenant.benchling.com'
  benchling_access_token: 'https://mytenant.benchling.com/api/v2/token'
  app_client_id: 'your-app-client-id'
  app_client_secret: 'your-app-client-secret'
AssaySchema:
  schema_id: "schema_api_id"
Project:
  project_id: "project_api_id"

The following is an example of the use of the assay_results_ingestion workflow.

import yaml
from mgtx_benchling_wrapper import (
    BenchlingContext,
    BenchlingWrapperFacade,
    AssayResultIngestionWorkflow
)
from mgtx_benchling_wrapper.utils.logger import get_logger

logger = get_logger(__name__,
                        file_log_level='DEBUG',
                        console_log_level='INFO', )

def config():
    with open("tests/config/test_config.yml") as f:
        return yaml.safe_load(f)

#create the benchling context        
ctx = BenchlingContext(
        base_url=config()['BenchlingCredentials']['benchling_url'],
        client_id=config()['BenchlingCredentials']['app_client_id'],
        client_secret=config()['BenchlingCredentials']['app_client_secret'],
        token_url=config()['BenchlingCredentials']['benchling_access_token'],
    )    

#initialize the wrapper
wrapper = BenchlingWrapperFacade(ctx.benchling())

#retrieve assay_schema_id
schema_id = config()['AssaySchema']['schema_id']

#retrieve project_id
project_id = config()['Project']['project_id']

#initiate results ingestion workflow
results_ingestion = AssayResultIngestionWorkflow(wrapper)

#ingest results on benchling
list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id']
        )       

The assay_results_ingestion function from the AssayResultIngestionWorkflow class above has two main ways of operating. The first part of the function focuses on how the results are ingested and the second part focuses on where the results are ingested.

Behavior 1: How the results are ingested 🔎

Let's first have a look at three possibilities on how the results are ingested. This behavior is ruled by the function inputs unique_identifiers, compare_on, and archive. The unique_identifiers list is used to call all assay results containing entities in this list. The compare_on list is used to build a dataframe of pre-existing results on Benchling that is used to subset rows not yet on Benchling and therefore need to be ingested.

We will consider a dummy results schema exists on Benchling with the following data:

We consider that we have a dataframe which results we want to ingest as follows:

Example 1 🚦

You want to archive pre-existing assay results and ingest results. In this specific case, data had already been ingested but was erroneously processed, so old values need to be archived and new values are to be ingested. To achieve this you could use the following piece of code

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = True
        compare_on = None
        )  

In this example, the following results would be archived:

and the following results would be ingested:

NOTE: You could keep the unique_identifiers = ['assay_run_id', 'sample_id'] in the code above to achieve the same result.

Example 2 🚦

You want to ingest values not yet on Benchling. You want to use a specific columns to make the comparison.

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = False
        compare_on = ['sample_id', 'replicate', 'average_result', 'result_sd']
        )  

In this example, no results would be archived and the following results would be ingested:

NOTE: You could keep the unique_identifiers = ['assay_run_id', 'sample_id'] in the code above to achieve the same result.

Example 3 🚦

You want to ingest values for entities that are not yet present on the result schema. This method is quite useful when ingesting high volumes of data in the background such as bioreactor online traces, where you don't want to lose time calling a big number of assay results.

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = False
        compare_on = None
        )  

In this example no results would be archived and no results would be ingested because all assay_run_id entities are already present in the schema.

NOTE: Best to use this method with setting the unique_identifiers to a list of one variable that represents the overarching batch run or assay run. In the case where unique_identifiers=['assay_run_id', 'sample_id'], if a new assay_run_id were to be ingested and a sample_id previously ingested was also mentioned in the dataframe, no results would be ingested.

Behavior 2: Where the results are ingested 🔎

There are two options: sending results to the warehouse only (backend) or sending results to a notebook entry (backend and frontend). Choose the appropriate method taking into account the volume of data you will be ingesting to minimize API calls and avoid hitting rate limits.

Let's use example 1 from section Behavior 1: How the results are ingested 🔎 as the basis.

Example 1: Sending results to warehouse only using transactions 🚦

To achieve this we would use the following code:

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = True,
        compare_on = None,
        entry_name = None,
        commit_in_transaction = True
        )  

This piece of code will ingest the results to the backend using transactions. Each transaction will contain requests with a maximum of 100 assay results. Transactions is the safest way to ingest data to backend given that when one ingestion in the transaction fails, all fail.

Example 2: Sending results to warehouse only using a bulk request 🚦

To achieve this we would use the following code:

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = True,
        compare_on = None,
        entry_name = None,
        commit_in_transaction = False
        )  

This piece of code will ingest the results to the backend using a bulk request. Up to 4000 assay runs can be ingested using the bulk request end-point. This is not as safe as a transaction but faster.

Example 3: Sending results to a notebook entry🚦

To achieve this we would use the following code:

list_missing_entities = results_ingestion.assay_results_ingestion(
        [dataframe_to_ingest],
        schema_id,
        project_id,
        unique_identifiers =['assay_run_id'],
        archive = True,
        compare_on = None,
        entry_name = 'Entry name containing a target table with of the schema_id type',
        commit_in_transaction = False
        )  

This piece of code will ingest the results to the entry with entry_name if a suited table is present. A table in a notebook entry is able to show up to 1000 values. Once results are ingested, navigate to the entry and click on the green flashing API button to see the data.

Any of the behaviors in examples on section Behavior 1: How the results are ingested 🔎 can be coupled with the behaviors in this section.

Creating a schema_definition

Another example is using the SchemaHandler to get a schema_definition the describes a schema on Benchling.

#run the following after initializing the wrapper

from mgtx_benchling_wrapper import SchemaHandler

schema_handler = SchemaHandler(wrapper)

schema_definition = schema_handler.build_schema_definition(schema_id, 'assay_results_schema')

Using a method directly from the wrapper

Another example is using a method from the wrapper directly to build your own workflow.

#run the following after initializing the wrapper

list_custom_entities = wrapper.custom_entities.get_by_names(['your-entity-1', 'your-entity-2'])

for custom_entity in list_custom_entities:
    name = wrapper.custom_entities.name()
    print(f"the name of the custom entity {custom_entity} is {name}.")

To release and publish run the following on gitBash to initialize the GitHub Action for publishing the package.

git tag v0.1.#
git push origin v0.1.#

Logs are by default enabled for developer use on the console. To save logs, an environment variable is needed. Create the environment variable on your repo in a .env file as such, and then make sure to load it.

APP_DEBUG_LOGS=1