Variants

class genomcore.controllers.variants.VariantsController(*args, **kwargs)

Bases: BaseController

Variant controller to interact with variants API. Variants API docs: https://variant-api.genomcore.net/api/docs

is_alive() str
get_template() Dict[str, Any]

Get the template for genetic variant observations in a project.

Returns:

template of a genetic variant observation.

Return type:

dict

create_observations(observations: List[dict], chunksize: int | None = 100000, retries: int | None = 0, measure_time: bool | None = False, chunk_details: bool | None = False) int

Save many genetic variant observations.

Parameters:
  • observations (list) – list of observations to save.

  • chunksize (int, optional) – size of the chunks to send. Defaults to 100000.

  • retries (int, optional) – number of retries in case of failure. Defaults to 3.

  • measure_time (bool, optional) – measure the time of each request. Defaults to False.

  • chunk_details (bool, optional) – log details of each chunk. Defaults to False.

Returns:

count of observations saved.

Return type:

int

Note

The ‘observations’ list should contain dictionaries with the following keys:

  • uri (str) Specific identifier for the observation. User provided.

[SNV_INDEL::hg19::chr2::58118555::58118555::A::G::NM_004006.2] - origin (str) The origin of the observation [GERMLINE or SOMATIC] - type (str) Type of observation [SNV_INDEL, CNV_SV, STRUCTURAL] - collection (str) The collection by which observations can be grouped together. Usually related to an study. User provided. [DemoStudy]

And any other required key-value pair that is part of the template.

Example:

from genomcore.client import Genomcore

api = Genomcore
observations = [
    {
        "uri": "TEST::VARIANT::URI::multi_e2e_1",
        "origin": "GERMLINE",
        "type": "SNV/INDEL",
        "collection": "e2eVarIndexID",
        "Position": {
            "VCF_Genome": "hg38",
            "VCF_Chr": "1",
            "VCF_Position": 1000,
            "VCF_ID": "rs1000"
        },
        "Genotype": {
            "VCF_Allele_REF": "A",
            "VCF_Allele_ALT": "G",
            "INFO_CALC_Genotype": "0/1"
        },
        "Calling_statistics": {
            "VCF_Filter": "PASS",
            "VCF_Quality": 80.0
        },
        "Gene": {
            "INFO_CSQ_SYMBOL": "GENE1"
        },
        "Feature": {
            "INFO_CSQ_Consequence": "missense_variant",
            "INFO_CSQ_Feature": "NR_10000",
            "INFO_CSQ_STRAND": "forward"
        }
    },
    ...
]

api.variant.create_observations(observations)
query_observations(page: int | None = 0, pageSize: int | None = 10000, body: List[dict] = None, measure_time: bool | None = False)

Get variant observations by providing MongoDB queries.

Parameters:
  • page (int, Optional) – Page number to query. Defaults to first page (0).

  • pageSize (int, Optional) – Number of entries per page. Defaults to 10000.

  • body (List[dict]) – Payload of the request containing MongoDB queries.

Returns:

List[dict] object from the query.

Example:

from genomcore.client import Genomcore

api = Genomcore(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")

# A valid body query in MongoDB format
# This example queries the time-series observations of the collection "TestBatch"
# Starting with page 2 of a page size of 500 entries per page.
body = {
    "filter": {
        "collection": "TestBatch"
    }
}
api.variant.query_observations(page = 2, pageSize = 500, body = body)
delete_observations(uri: str | None = None, collection: str | None = None, measure_time: bool | None = False) int

Delete variant observations.

Parameters:
  • uri (str) – observation URI

  • collection (str) – collection that aggregates multiple observations

Returns:

count of deleted observations

Return type:

int

Warning

At least one of the arguments ‘uri’ or ‘collection’ must be provided.

Example:

from genomcore.client import Genomcore

api = Genomcore(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")

api.variant.delete_observations(uri="TEST::VARIANT::URI::multi_e2e_1")
count(filter={})

count observations.

Parameters:

filter (dict) – Payload of the request containing MongoDB queries

Returns:

count observations

Return type:

int

__annotations__ = {}
__firstlineno__ = 13
__static_attributes__ = ('_requester',)