Module phc.easy.omics.genomic_test

Classes

class GenomicTest

Expand source code

class GenomicTest(PagingApiItem):
    @staticmethod
    def resource_path():
        return "genomics/projects/{project_id}/tests"

    @staticmethod
    def params_class():
        return GenomicTestOptions

    @staticmethod
    def transform_results(
        data_frame: pd.DataFrame, params: dict, **expand_args
    ):
        args = {
            **expand_args,
            "code_columns": [
                *expand_args.get("code_columns", []),
                "bodySite",
                "patient",
            ],
            "custom_columns": [
                *expand_args.get("custom_columns", []),
                Frame.codeable_like_column_expander("sourceFile"),
            ],
        }
        df = Frame.expand(data_frame, **args)

        if "sets" in df.columns:
            df = (
                pd.concat(
                    data_frame.apply(
                        lambda x: pd.DataFrame(
                            [{"index": x.name, **s} for s in x.sets]
                        ),
                        axis=1,
                    ).values
                )
                .join(df.drop(["sets"], axis=1), on="index", rsuffix=".test")
                .drop(["index"], axis=1)
                .reset_index(drop=True)
            )

        test_type = params.get("type", None)

        if test_type and len(df) > 0:
            # TODO: Remove when API fixed

            # NOTE: The API does not filter the returned sets because it is a
            # nested structure. Since it's not a boatload of information, we opt
            # to filter client-side for now.
            return df[df.setType == test_type].reset_index(drop=True)

        return df

    @classmethod
    def get_data_frame(
        cls,
        patient_id: Optional[str] = None,
        status: Optional[GenomicTestStatus] = GenomicTestStatus.ACTIVE,
        test_type: Optional[GenomicTestType] = None,
        all_results: bool = False,
        auth_args: Auth = Auth.shared(),
        max_pages: Optional[int] = None,
        page_size: Optional[int] = None,
        log: bool = False,
        ignore_cache: bool = False,
        **kw_args,
    ):
        """Execute a request for genomic tests

        ## Parameters

        Query: `phc.easy.omics.options.genomic_test.GenomicTestOptions`

        Execution: `phc.easy.query.Query.execute_paging_api`

        Expansion: `phc.easy.frame.Frame.expand`

        NOTE: `test_type` is translated to `type` as a parameter
        """
        df = super().get_data_frame(
            **kw_args, **cls._get_current_args(inspect.currentframe(), locals())
        )

        return df

Ancestors

• PagingApiItem

Static methods

def get_data_frame(patient_id: Optional[str] = None, status: Optional[GenomicTestStatus] = GenomicTestStatus.ACTIVE, test_type: Optional[GenomicTestType] = None, all_results: bool = False, auth_args: Auth = <phc.easy.auth.Auth object>, max_pages: Optional[int] = None, page_size: Optional[int] = None, log: bool = False, ignore_cache: bool = False, **kw_args)

Execute a request for genomic tests

Parameters

Query: phc.easy.omics.options.genomic_test.GenomicTestOptions

Execution: Query.execute_paging_api()

Expansion: Frame.expand()

NOTE: test_type is translated to type as a parameter

Expand source code

@classmethod
def get_data_frame(
    cls,
    patient_id: Optional[str] = None,
    status: Optional[GenomicTestStatus] = GenomicTestStatus.ACTIVE,
    test_type: Optional[GenomicTestType] = None,
    all_results: bool = False,
    auth_args: Auth = Auth.shared(),
    max_pages: Optional[int] = None,
    page_size: Optional[int] = None,
    log: bool = False,
    ignore_cache: bool = False,
    **kw_args,
):
    """Execute a request for genomic tests

    ## Parameters

    Query: `phc.easy.omics.options.genomic_test.GenomicTestOptions`

    Execution: `phc.easy.query.Query.execute_paging_api`

    Expansion: `phc.easy.frame.Frame.expand`

    NOTE: `test_type` is translated to `type` as a parameter
    """
    df = super().get_data_frame(
        **kw_args, **cls._get_current_args(inspect.currentframe(), locals())
    )

    return df

def params_class()

Inherited from: PagingApiItem.params_class

Returns a pydantic type that validates and transforms the params with dict()

Expand source code

@staticmethod
def params_class():
    return GenomicTestOptions

def process_params(params: dict) ‑> dict

Inherited from: PagingApiItem.process_params

Validates and transforms the API query parameters

def resource_path()

Inherited from: PagingApiItem.resource_path

Returns the API url name for retrieval

Expand source code

@staticmethod
def resource_path():
    return "genomics/projects/{project_id}/tests"

def transform_results(data_frame: pandas.core.frame.DataFrame, params: dict, **expand_args)

Inherited from: PagingApiItem.transform_results

Transform data frame batch

Expand source code

@staticmethod
def transform_results(
    data_frame: pd.DataFrame, params: dict, **expand_args
):
    args = {
        **expand_args,
        "code_columns": [
            *expand_args.get("code_columns", []),
            "bodySite",
            "patient",
        ],
        "custom_columns": [
            *expand_args.get("custom_columns", []),
            Frame.codeable_like_column_expander("sourceFile"),
        ],
    }
    df = Frame.expand(data_frame, **args)

    if "sets" in df.columns:
        df = (
            pd.concat(
                data_frame.apply(
                    lambda x: pd.DataFrame(
                        [{"index": x.name, **s} for s in x.sets]
                    ),
                    axis=1,
                ).values
            )
            .join(df.drop(["sets"], axis=1), on="index", rsuffix=".test")
            .drop(["index"], axis=1)
            .reset_index(drop=True)
        )

    test_type = params.get("type", None)

    if test_type and len(df) > 0:
        # TODO: Remove when API fixed

        # NOTE: The API does not filter the returned sets because it is a
        # nested structure. Since it's not a boatload of information, we opt
        # to filter client-side for now.
        return df[df.setType == test_type].reset_index(drop=True)

    return df

class GenomicTestOptions (**data: Any)

Options to pass to /v1/genomics/projects/{project_id}/tests

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Expand source code

class GenomicTestOptions(PagingApiOptions):
    """Options to pass to `/v1/genomics/projects/{project_id}/tests`"""

    patient_id: Optional[str]
    status: Optional[GenomicTestStatus]
    test_type: Optional[GenomicTestType]

    @staticmethod
    def transform(key, value):
        new_key = {"patient_id": "patientId", "test_type": "type"}.get(key, key)

        return (new_key, value)

Ancestors

• PagingApiOptions
• pydantic.main.BaseModel

Class variables

var model_config

var patient_id : Optional[str]

var status : Optional[GenomicTestStatus]

var test_type : Optional[GenomicTestType]

Static methods

def transform(key, value)

Expand source code

@staticmethod
def transform(key, value):
    new_key = {"patient_id": "patientId", "test_type": "type"}.get(key, key)

    return (new_key, value)

Methods

def model_dump(self)

Inherited from: PagingApiOptions.model_dump

Usage docs: https://docs.pydantic.dev/2.10/concepts/serialization/#modelmodel_dump …