Metadata-Version: 2.4
Name: mgtx-benchling-wrapper
Version: 0.1.8
Summary: Python wrapper for Benchling API with common functions used at MGTX DSC
Author-email: Ana Valinhas <ana.valinhas@meiragtx.com>
License: MIT
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: benchling-sdk>=1.21.2
Requires-Dist: benchling-api-client>=2.0.342
Requires-Dist: pandas>=2.2.2
Requires-Dist: numpy>=1.26.4
Provides-Extra: dev
Requires-Dist: pytest>=9.0.3; extra == "dev"
Requires-Dist: pytest-mock>=3.15.1; extra == "dev"
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"
Dynamic: license-file

# 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
```
