Skip to content

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
133
134
135
136
137
138
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
248
249
250
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
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
@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):  # noqa: PLR6301
    """
    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
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
@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
328
329
330
331
332
333
334
335
336
337
338
339
340
341
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
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
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
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def create_analysis(  # noqa: PLR0917
    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}, 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
436
437
438
439
440
441
442
443
444
445
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
447
448
449
450
451
452
453
454
455
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

parse staticmethod

parse(name)

Parse str and create a AnalysisStatus object

Source code in src/cpg_flow/metamist.py
159
160
161
162
163
164
@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.

parse staticmethod

parse(val)

Parse str and create a AnalysisStatus object

Source code in src/cpg_flow/metamist.py
187
188
189
190
191
192
193
194
195
196
197
@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

parse staticmethod

parse(data)

Parse data to create an Analysis object.

Source code in src/cpg_flow/metamist.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
@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

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
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
@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
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
def get_cohort_sgs(cohort_id: str) -> dict:
    """
    Retrieve sequencing group entries for a single cohort.
    """
    logger.info(f'Getting sequencing groups for cohort {cohort_id}')
    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"
    #     "dataset": "DatasetName"
    # }
    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']
    cohort_dataset = entries['cohorts'][0]['project']['dataset']
    sequencing_groups = entries['cohorts'][0]['sequencingGroups']

    return {
        'name': cohort_name,
        'dataset': cohort_dataset,
        '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
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
def parse_reads(
    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',
        )

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

    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 {sorted(supported_types)}',
        )

    if reads_type in {'bam', 'cram'}:
        return find_cram_or_bam(
            reads_data,
            sequencing_group_id,
            check_existence,
            reference_assembly,
            access_level=get_config()['workflow']['access_level'],
        )
    else:
        return find_fastqs(reads_data, sequencing_group_id, check_existence)