mgtx-benchling-wrapper 0.1.8

Python wrapper for Benchling API with common functions used at MGTX DSC

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:

assay_run_id	sample_id	replicate	average_result	result_sd	valid?
Assay 1	Sample 1	1	25	2	Yes
Assay 1	Sample 1	2	34	6	Yes
Assay 2	Sample 1	1	15	4	Yes
Assay 2	Sample 2	1	31	5	Yes

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

assay_run_id	sample_id	replicate	average_result	result_sd	valid?
Assay 2	Sample 1	1	31	5	Yes
Assay 2	Sample 2	1	50	8	Yes

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:

assay_run_id	sample_id	replicate	average_result	result_sd	valid?
Assay 2	Sample 1	1	15	4	Yes
Assay 2	Sample 2	1	31	5	Yes

and the following results would be ingested:

assay_run_id	sample_id	replicate	average_result	result_sd	valid?
Assay 2	Sample 1	1	31	5	Yes
Assay 2	Sample 2	1	50	8	Yes

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:

assay_run_id	sample_id	replicate	average_result	result_sd	valid?
Assay 2	Sample 1	1	31	5	Yes
Assay 2	Sample 2	1	50	8	Yes

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).

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

Example 1: Sending results to warehouse 🚦

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 and you will be able to visualize them in the entity directly under the results tab.

Example 2: 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. 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