Gene index

@dataclass
class GeneIndexStep(GeneIndexStepConfig):
    """Gene index step.

    This step generates a gene index dataset from an Open Targets Platform target dataset.
    """

    session: Session = Session()

    def run(self: GeneIndexStep) -> None:
        """Run Target index step."""
        # Extract
        platform_target = self.session.spark.read.parquet(self.target_path)
        # Transform
        gene_index = GeneIndex.from_source(platform_target)
        # Load
        gene_index.df.write.mode(self.session.write_mode).parquet(self.gene_index_path)

@dataclass
class GeneIndexStepConfig:
    """Gene index step requirements.

    Attributes:
        target_path (str): Open targets Platform target dataset path.
        gene_index_path (str): Output gene index path.
    """

    _target_: str = "otg.gene_index.GeneIndexStep"
    target_path: str = MISSING
    gene_index_path: str = MISSING