Metamist

Helpers to communicate with the metamist database.

cpg_flow.metamist.get_metamist

get_metamist()

Return the cohort object

Source code in src/cpg_flow/metamist.py

def get_metamist() -> 'Metamist':
    """Return the cohort object"""
    global _metamist
    if not _metamist:
        _metamist = Metamist()
    return _metamist

cpg_flow.metamist.Metamist

Metamist()

Communication with metamist.

Source code in src/cpg_flow/metamist.py

def __init__(self) -> None:
    self.default_dataset: str = get_config()['workflow']['dataset']
    self.aapi = AnalysisApi()

default_dataset instance-attribute

default_dataset = get_config()['workflow']['dataset']

aapi instance-attribute

aapi = AnalysisApi()

make_retry_aapi_call

make_retry_aapi_call(api_func, **kwargv)

Make a generic API call to self.aapi with retries. Retry only if ServiceException is thrown

TODO: How many retries? e.g. try 3 times, wait 2^3: 8, 16, 24 seconds

Source code in src/cpg_flow/metamist.py

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=3, min=8, max=30),
    retry=retry_if_exception_type(ServiceException),
    reraise=True,
)
def make_retry_aapi_call(self, api_func: Callable, **kwargv: Any):
    """
    Make a generic API call to self.aapi with retries.
    Retry only if ServiceException is thrown

    TODO: How many retries?
    e.g. try 3 times, wait 2^3: 8, 16, 24 seconds
    """
    try:
        return api_func(**kwargv)
    except ServiceException:
        # raise here so the retry occurs
        LOGGER.warning(
            f'Retrying {api_func} ...',
        )
        raise

make_aapi_call

make_aapi_call(api_func, **kwargv)

Make a generic API call to self.aapi. This is a wrapper around retry of API call to handle exceptions and LOGGER.

Source code in src/cpg_flow/metamist.py

def make_aapi_call(self, api_func: Callable, **kwargv: Any):
    """
    Make a generic API call to self.aapi.
    This is a wrapper around retry of API call to handle exceptions and LOGGER.
    """
    try:
        return self.make_retry_aapi_call(api_func, **kwargv)
    except (ServiceException, ApiException) as e:
        # Metamist API failed even after retries
        # log the error and continue
        traceback.print_exc()
        LOGGER.error(
            f'Error: {e} Call {api_func} failed with payload:\n{kwargv!s}',
        )
    # TODO: discuss should we catch all here as well?
    # except Exception as e:
    #     # Other exceptions?

    return None

get_sg_entries

get_sg_entries(dataset_name)

Retrieve sequencing group entries for a dataset, in the context of access level and filtering options.

Source code in src/cpg_flow/metamist.py

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=3, min=8, max=30),
    retry=retry_if_exception_type(TransportServerError),
    reraise=True,
)
def get_sg_entries(self, dataset_name: str) -> list[dict]:
    """
    Retrieve sequencing group entries for a dataset, in the context of access level
    and filtering options.
    """
    metamist_proj = self.get_metamist_proj(dataset_name)
    LOGGER.info(f'Getting sequencing groups for dataset {metamist_proj}')

    skip_sgs = get_config()['workflow'].get('skip_sgs', [])
    only_sgs = get_config()['workflow'].get('only_sgs', [])
    sequencing_type = get_config()['workflow'].get('sequencing_type')

    if only_sgs and skip_sgs:
        raise MetamistError('Cannot specify both only_sgs and skip_sgs in config')

    sequencing_group_entries = query(
        GET_SEQUENCING_GROUPS_QUERY,
        variables={
            'metamist_proj': metamist_proj,
            'only_sgs': only_sgs,
            'skip_sgs': skip_sgs,
            'sequencing_type': sequencing_type,
        },
    )

    return sequencing_group_entries['project']['sequencingGroups']

update_analysis

update_analysis(analysis, status)

Update "status" of an Analysis entry.

Source code in src/cpg_flow/metamist.py

def update_analysis(self, analysis: Analysis, status: AnalysisStatus):
    """
    Update "status" of an Analysis entry.
    """
    self.make_aapi_call(
        self.aapi.update_analysis,
        analysis_id=analysis.id,
        analysis_update_model=models.AnalysisUpdateModel(
            status=models.AnalysisStatus(status.value),
        ),
    )
    # Keeping this as is for compatibility with the existing code
    # However this should only be set after the API call is successful
    analysis.status = status

get_analyses_by_sgid

get_analyses_by_sgid(
    analysis_type, analysis_status=COMPLETED, dataset=None
)

Query the DB to find the last completed analysis for the type, sequencing group ids, and sequencing type, one Analysis object per sequencing group. Assumes the analysis is defined for a single sequencing group (that is, analysis_type=cram|gvcf|qc).

Source code in src/cpg_flow/metamist.py

def get_analyses_by_sgid(
    self,
    analysis_type: AnalysisType,
    analysis_status: AnalysisStatus = AnalysisStatus.COMPLETED,
    dataset: str | None = None,
) -> dict[str, Analysis]:
    """
    Query the DB to find the last completed analysis for the type, sequencing group ids,
    and sequencing type, one Analysis object per sequencing group. Assumes the analysis
    is defined for a single sequencing group (that is, analysis_type=cram|gvcf|qc).
    """
    metamist_proj = self.get_metamist_proj(dataset)

    analyses = query(
        GET_ANALYSES_QUERY,
        variables={
            'metamist_proj': metamist_proj,
            'analysis_type': analysis_type.value,
            'analysis_status': analysis_status.name,
        },
    )

    analysis_per_sid: dict[str, Analysis] = dict()

    for analysis in analyses['project']['analyses']:
        a = Analysis.parse(analysis)
        if not a:
            continue

        assert a.status == analysis_status, analysis
        assert a.type == analysis_type, analysis
        if len(a.sequencing_group_ids) < 1:
            LOGGER.warning(f'Analysis has no sequencing group ids. {analysis}')
            continue

        assert len(a.sequencing_group_ids) == 1, analysis
        analysis_per_sid[list(a.sequencing_group_ids)[0]] = a

    LOGGER.info(
        f'Querying {analysis_type} analysis entries for {metamist_proj}: found {len(analysis_per_sid)}',
    )
    return analysis_per_sid

create_analysis

create_analysis(
    output,
    type_,
    status,
    cohort_ids=None,
    sequencing_group_ids=None,
    dataset=None,
    meta=None,
)

Tries to create an Analysis entry, returns its id if successful.

Source code in src/cpg_flow/metamist.py

def create_analysis(
    self,
    output: Path | str,
    type_: str | AnalysisType,
    status: str | AnalysisStatus,
    cohort_ids: list[str] | None = None,
    sequencing_group_ids: list[str] | None = None,
    dataset: str | None = None,
    meta: dict | None = None,
) -> int | None:
    """
    Tries to create an Analysis entry, returns its id if successful.
    """
    metamist_proj = self.get_metamist_proj(dataset)

    if isinstance(type_, AnalysisType):
        type_ = type_.value
    if isinstance(status, AnalysisStatus):
        status = status.value

    if not cohort_ids:
        cohort_ids = []

    if not sequencing_group_ids:
        sequencing_group_ids = []

    am = models.Analysis(
        type=type_,
        status=models.AnalysisStatus(status),
        output=str(output),
        cohort_ids=list(cohort_ids),
        sequencing_group_ids=list(sequencing_group_ids),
        meta=meta or {},
    )
    aid = self.make_aapi_call(
        self.aapi.create_analysis,
        project=metamist_proj,
        analysis=am,
    )
    if aid is None:
        LOGGER.error(
            f'Failed to create Analysis(type={type_}, status={status}, output={output!s}) in {metamist_proj}',
        )
        return None
    LOGGER.info(
        f'Created Analysis(id={aid}, type={type_}, status={status}, ' f'output={output!s}) in {metamist_proj}',
    )
    return aid

get_ped_entries

get_ped_entries(dataset=None)

Retrieve PED lines for a specified SM project, with external participant IDs.

Source code in src/cpg_flow/metamist.py

def get_ped_entries(self, dataset: str | None = None) -> list[dict[str, str]]:
    """
    Retrieve PED lines for a specified SM project, with external participant IDs.
    """
    metamist_proj = self.get_metamist_proj(dataset)
    entries = query(GET_PEDIGREE_QUERY, variables={'metamist_proj': metamist_proj})

    pedigree_entries = entries['project']['pedigree']

    return pedigree_entries

get_metamist_proj

get_metamist_proj(dataset=None)

Return the Metamist project name, appending '-test' if the access level is 'test'.

Source code in src/cpg_flow/metamist.py

def get_metamist_proj(self, dataset: str | None = None) -> str:
    """
    Return the Metamist project name, appending '-test' if the access level is 'test'.
    """
    metamist_proj = dataset or self.default_dataset
    if get_config()['workflow']['access_level'] == 'test' and not metamist_proj.endswith('-test'):
        metamist_proj += '-test'

    return metamist_proj

cpg_flow.metamist.AnalysisStatus

Bases: Enum

Corresponds to metamist Analysis statuses: https://github.com/populationgenomics/sample-metadata/blob/dev/models/enums/analysis.py#L14-L21

QUEUED class-attribute instance-attribute

QUEUED = 'queued'

IN_PROGRESS class-attribute instance-attribute

IN_PROGRESS = 'in-progress'

FAILED class-attribute instance-attribute

FAILED = 'failed'

COMPLETED class-attribute instance-attribute

COMPLETED = 'completed'

UNKNOWN class-attribute instance-attribute

UNKNOWN = 'unknown'

parse staticmethod

parse(name)

Parse str and create a AnalysisStatus object

Source code in src/cpg_flow/metamist.py

@staticmethod
def parse(name: str) -> 'AnalysisStatus':
    """
    Parse str and create a AnalysisStatus object
    """
    return {v.value: v for v in AnalysisStatus}[name.lower()]

cpg_flow.metamist.AnalysisType

Bases: Enum

Corresponds to metamist Analysis types: https://github.com/populationgenomics/sample-metadata/blob/dev/models/enums /analysis.py#L4-L11

Re-defined in a separate module to decouple from the main metamist module, so decorators can use @stage(analysis_type=AnalysisType.QC) without importing the metamist package.

QC class-attribute instance-attribute

QC = 'qc'

JOINT_CALLING class-attribute instance-attribute

JOINT_CALLING = 'joint-calling'

GVCF class-attribute instance-attribute

GVCF = 'gvcf'

CRAM class-attribute instance-attribute

CRAM = 'cram'

MITO_CRAM class-attribute instance-attribute

MITO_CRAM = 'mito-cram'

CUSTOM class-attribute instance-attribute

CUSTOM = 'custom'

ES_INDEX class-attribute instance-attribute

ES_INDEX = 'es-index'

COMBINER class-attribute instance-attribute

COMBINER = 'combiner'

parse staticmethod

parse(val)

Parse str and create a AnalysisStatus object

Source code in src/cpg_flow/metamist.py

@staticmethod
def parse(val: str) -> 'AnalysisType':
    """
    Parse str and create a AnalysisStatus object
    """
    d = {v.value: v for v in AnalysisType}
    if val not in d:
        raise MetamistError(
            f'Unrecognised analysis type {val}. Available: {list(d.keys())}',
        )
    return d[val.lower()]

cpg_flow.metamist.Analysis dataclass

Analysis(
    id, type, status, sequencing_group_ids, output, meta
)

Metamist DB Analysis entry.

See the metamist package for more details: https://github.com/populationgenomics/sample-metadata

id instance-attribute

id

type instance-attribute

type

status instance-attribute

status

sequencing_group_ids instance-attribute

sequencing_group_ids

output instance-attribute

output

meta instance-attribute

meta

parse staticmethod

parse(data)

Parse data to create an Analysis object.

Source code in src/cpg_flow/metamist.py

@staticmethod
def parse(data: dict) -> 'Analysis':
    """
    Parse data to create an Analysis object.
    """
    req_keys = ['id', 'type', 'status']
    if any(k not in data for k in req_keys):
        for key in req_keys:
            if key not in data:
                LOGGER.error(f'"Analysis" data does not have {key}: {data}')
        raise ValueError(f'Cannot parse metamist Sequence {data}')

    output = data.get('output')
    if output:
        output = to_path(output)

    a = Analysis(
        id=int(data['id']),
        type=AnalysisType.parse(data['type']),
        status=AnalysisStatus.parse(data['status']),
        sequencing_group_ids=set([s['id'] for s in data['sequencingGroups']]),
        output=output,
        meta=data.get('meta') or {},
    )
    return a

cpg_flow.metamist.Assay dataclass

Assay(
    id,
    sequencing_group_id,
    meta,
    assay_type,
    alignment_input=None,
)

Metamist "Assay" entry.

See metamist for more details: https://github.com/populationgenomics/sample-metadata

id instance-attribute

id

sequencing_group_id instance-attribute

sequencing_group_id

meta instance-attribute

meta

assay_type instance-attribute

assay_type

alignment_input class-attribute instance-attribute

alignment_input = None

parse staticmethod

parse(
    data, sg_id, check_existence=False, run_parse_reads=True
)

Create from a dictionary.

Source code in src/cpg_flow/metamist.py

@staticmethod
def parse(
    data: dict,
    sg_id: str,
    check_existence: bool = False,
    run_parse_reads: bool = True,
) -> 'Assay':
    """
    Create from a dictionary.
    """

    assay_keys = ['id', 'type', 'meta']
    missing_keys = [key for key in assay_keys if data.get(key) is None]

    if missing_keys:
        raise ValueError(
            f'Cannot parse metamist Sequence {data}. Missing keys: {missing_keys}',
        )

    assay_type = str(data['type'])
    assert assay_type, data
    mm_seq = Assay(
        id=str(data['id']),
        sequencing_group_id=sg_id,
        meta=data['meta'],
        assay_type=assay_type,
    )
    if run_parse_reads:
        mm_seq.alignment_input = parse_reads(
            sequencing_group_id=sg_id,
            assay_meta=data['meta'],
            check_existence=check_existence,
        )
    return mm_seq

cpg_flow.metamist.get_cohort_sgs

get_cohort_sgs(cohort_id)

Retrieve sequencing group entries for a single cohort.

Source code in src/cpg_flow/metamist.py

def get_cohort_sgs(cohort_id: str) -> dict:
    """
    Retrieve sequencing group entries for a single cohort.
    """
    entries = query(GET_SEQUENCING_GROUPS_BY_COHORT_QUERY, {'cohort_id': cohort_id})

    # Create dictionary keying sequencing groups by project and including cohort name
    # {
    #     "sequencing_groups": {
    #         project_id: [sequencing_group_1, sequencing_group_2, ...],
    #         ...
    #     },
    #     "name": "CohortName"
    # }
    if len(entries['cohorts']) != 1:
        raise MetamistError('We only support one cohort at a time currently')

    if entries.get('data') is None and 'errors' in entries:
        message = entries['errors'][0]['message']
        raise MetamistError(f'Error fetching cohort: {message}')

    cohort_name = entries['cohorts'][0]['name']
    sequencing_groups = entries['cohorts'][0]['sequencingGroups']

    return {
        'name': cohort_name,
        'sequencing_groups': sequencing_groups,
    }

cpg_flow.metamist.parse_reads

parse_reads(
    sequencing_group_id, assay_meta, check_existence
)

Parse a AlignmentInput object from the meta dictionary. check_existence: check if fastq/crams exist on buckets. Default value is pulled from self.metamist and can be overridden.

Source code in src/cpg_flow/metamist.py

def parse_reads(  # pylint: disable=too-many-return-statements
    sequencing_group_id: str,
    assay_meta: dict,
    check_existence: bool,
) -> AlignmentInput:
    """
    Parse a AlignmentInput object from the meta dictionary.
    `check_existence`: check if fastq/crams exist on buckets.
    Default value is pulled from self.metamist and can be overridden.
    """
    reads_data = assay_meta.get('reads')
    reads_type = assay_meta.get('reads_type')
    reference_assembly = assay_meta.get('reference_assembly', {}).get('location')

    if not reads_data:
        raise MetamistError(f'{sequencing_group_id}: no "meta/reads" field in meta')
    if not reads_type:
        raise MetamistError(
            f'{sequencing_group_id}: no "meta/reads_type" field in meta',
        )
    supported_types = ('fastq', 'bam', 'cram')
    if reads_type not in supported_types:
        raise MetamistError(
            f'{sequencing_group_id}: ERROR: "reads_type" is expected to be one of {supported_types}',
        )

    if reads_type in ('bam', 'cram'):
        if len(reads_data) > 1:
            raise MetamistError(
                f'{sequencing_group_id}: supporting only single bam/cram input',
            )

        location = reads_data[0]['location']
        if not (location.endswith('.cram') or location.endswith('.bam')):
            raise MetamistError(
                f'{sequencing_group_id}: ERROR: expected the file to have an extension '
                f'.cram or .bam, got: {location}',
            )
        if get_config()['workflow']['access_level'] == 'test':
            location = location.replace('-main-upload/', '-test-upload/')
        if check_existence and not exists(location):
            raise MetamistError(
                f'{sequencing_group_id}: ERROR: index file does not exist: {location}',
            )

        # Index:
        index_location = None
        if reads_data[0].get('secondaryFiles'):
            index_location = reads_data[0]['secondaryFiles'][0]['location']
            if (location.endswith('.cram') and not index_location.endswith('.crai')) or (
                location.endswith('.bam') and not index_location.endswith('.bai')
            ):
                raise MetamistError(
                    f'{sequencing_group_id}: ERROR: expected the index file to have an extension '
                    f'.crai or .bai, got: {index_location}',
                )
            if get_config()['workflow']['access_level'] == 'test':
                index_location = index_location.replace(
                    '-main-upload/',
                    '-test-upload/',
                )
            if check_existence and not exists(index_location):
                raise MetamistError(
                    f'{sequencing_group_id}: ERROR: index file does not exist: {index_location}',
                )

        if location.endswith('.cram'):
            return CramPath(
                location,
                index_path=index_location,
                reference_assembly=reference_assembly,
            )
        assert location.endswith('.bam')
        return BamPath(location, index_path=index_location)

    fastq_pairs = FastqPairs()

    for lane_pair in reads_data:
        if len(lane_pair) != 2:
            raise ValueError(
                f'Sequence data for sequencing group {sequencing_group_id} is incorrectly '
                f'formatted. Expecting 2 entries per lane (R1 and R2 fastqs), '
                f'but got {len(lane_pair)}. '
                f'Read data: {pprint.pformat(lane_pair)}',
            )
        if check_existence and not exists(lane_pair[0]['location']):
            raise MetamistError(
                f'{sequencing_group_id}: ERROR: read 1 file does not exist: {lane_pair[0]["location"]}',
            )
        if check_existence and not exists(lane_pair[1]['location']):
            raise MetamistError(
                f'{sequencing_group_id}: ERROR: read 2 file does not exist: {lane_pair[1]["location"]}',
            )

        fastq_pairs.append(
            FastqPair(
                to_path(lane_pair[0]['location']),
                to_path(lane_pair[1]['location']),
            ),
        )

    return fastq_pairs