dae.query_variants.sql.schema2 package

Subpackages

Submodules

dae.query_variants.sql.schema2.base_query_builder module

class dae.query_variants.sql.schema2.base_query_builder.BaseQueryBuilder(dialect: Dialect, db: str | None, family_variant_table: str | None, summary_allele_table: str, pedigree_table: str, family_variant_schema: dict[str, str] | None, summary_allele_schema: dict[str, str], partition_config: dict[str, Any] | None, pedigree_schema: dict[str, str], families: FamiliesData, gene_models: GeneModels | None = None, reference_genome: ReferenceGenome | None = None)[source]

Bases: ABC

Class that abstracts away the process of building a query.

GENE_REGIONS_HEURISTIC_CUTOFF = 20
GENE_REGIONS_HEURISTIC_EXTEND = 20000
MAX_CHILD_NUMBER = 9999
QUOTE = "'"
WHERE = '\n  WHERE\n    {where}\n    '
build_query(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, family_ids: Iterable[str] | None = None, person_ids: Iterable[str] | None = None, inheritance: str | list[str] | None = None, roles: str | None = None, sexes: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, pedigree_fields: tuple | None = None) → str[source]

Build an SQL query in the correct order.

class dae.query_variants.sql.schema2.base_query_builder.Dialect(namespace: str | None = None)[source]

Bases: object

Caries info about a SQL dialect.

static add_unnest_in_join() → bool[source]
static array_item_suffix() → str[source]
build_array_join(column: str, allias: str) → str[source]
build_table_name(table: str, db: str | None) → str[source]
static escape_char() → str[source]
static escape_quote_char() → str[source]
static float_type() → str[source]
static int_type() → str[source]
static use_bit_and_function() → bool[source]

dae.query_variants.sql.schema2.base_variants module

class dae.query_variants.sql.schema2.base_variants.SqlSchema2Variants(dialect: Dialect, db: str | None, family_variant_table: str | None, summary_allele_table: str | None, pedigree_table: str, meta_table: str, gene_models: GeneModels | None = None)[source]

Bases: QueryVariantsBase

Base class for Schema2 SQL like variants’ query interface.

RUNNER_CLASS: type[QueryRunner]
build_family_variants_query_runner(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, family_ids: list[str] | None = None, person_ids: list[str] | None = None, inheritance: list[str] | None = None, roles: str | None = None, sexes: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, study_filters: list[str] | None = None, pedigree_fields: tuple | None = None, **kwargs: Any) → QueryRunner[source]

Build a query selecting the appropriate family variants.

static build_person_set_collection_query(person_set_collection: PersonSetCollection, person_set_collection_query: tuple[str, set[str]]) → tuple | tuple[list[str], list[str]] | None[source]

No idea what it does. If you know please edit.

build_summary_variants_query_runner(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, **kwargs: Any) → QueryRunner[source]

Build a query selecting the appropriate summary variants.

query_summary_variants(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, **kwargs: Any) → Generator[SummaryVariant, None, None][source]

Query summary variants.

query_variants(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, family_ids: list[str] | None = None, person_ids: list[str] | None = None, inheritance: list[str] | None = None, roles: str | None = None, sexes: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, pedigree_fields: tuple | None = None, **kwargs: Any) → Generator[FamilyVariant, None, None][source]

Query family variants.

dae.query_variants.sql.schema2.family_builder module

class dae.query_variants.sql.schema2.family_builder.FamilyQueryBuilder(dialect: Dialect, db: str, family_variant_table: str, summary_allele_table: str, pedigree_table: str, family_variant_schema: dict[str, str], summary_allele_schema: dict[str, str], table_properties: dict | None, pedigree_schema: dict[str, str], families: FamiliesData, gene_models: GeneModels | None = None, do_join_pedigree: bool = False, do_join_allele_in_members: bool = False)[source]

Bases: BaseQueryBuilder

Build queries related to family variants.

dae.query_variants.sql.schema2.sql_query_builder module

class dae.query_variants.sql.schema2.sql_query_builder.Db2Layout(db: str | None, study: str, pedigree: str, summary: str | None, family: str | None, meta: str)[source]

Bases: object

Genotype data layout in the database.

db: str | None
family: str | None
meta: str
pedigree: str
study: str
summary: str | None
class dae.query_variants.sql.schema2.sql_query_builder.QueryBuilderBase(schema: Schema, families: FamiliesData, partition_descriptor: PartitionDescriptor | None, gene_models: GeneModels, reference_genome: ReferenceGenome)[source]

Bases: object

Base class for building SQL queries.

GENE_REGIONS_HEURISTIC_CUTOFF = 20
GENE_REGIONS_HEURISTIC_EXTEND = 20000
REGION_BINS_HEURISTIC_CUTOFF = 20
all_region_bins() → list[str][source]

Return all region bins.

build_gene_regions(genes: list[str], regions: list[Region] | None) → list[Region] | None[source]

Build a list of regions based on genes.

static build_inheritance_query(inheritance_query: Sequence[str], attr: str) → str[source]

Construct an inheritance query.

static build_roles_query(roles_query: str, attr: str) → str[source]

Construct a roles query.

static build_schema(summary_schema: dict[str, str], family_schema: dict[str, str], pedigree_schema: dict[str, str]) → Schema[source]
static build_sexes_query(sexes_query: str, attr: str) → str[source]

Build sexes query.

static build_variant_types_query(variant_types_query: str, attr: str) → str[source]

Build a variant types query.

calc_batched_heuristics(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, inheritance: Sequence[str] | None = None, roles: str | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, family_ids: Iterable[str] | None = None, person_ids: Iterable[str] | None = None) → list[QueryHeuristics][source]

Calculate heuristics baches for a query.

calc_coding_bins(effect_types: Sequence[str] | None) → list[str][source]

Calculate applicable coding bins for a query.

calc_family_bins(family_ids: Iterable[str] | None, person_ids: Iterable[str] | None) → list[str][source]

Calculate family bins for a query.

calc_frequency_bins(*, inheritance: Sequence[str] | None = None, roles: str | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None) → list[str][source]

Calculate applicable frequency bins for a query.

calc_heuristics(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, inheritance: Sequence[str] | None = None, roles: str | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, family_ids: Iterable[str] | None = None, person_ids: Iterable[str] | None = None) → QueryHeuristics[source]

Calculate heuristic bins for a query.

calc_region_bins(regions: list[Region] | None) → list[str][source]

Calculate applicable region bins for a query.

static check_inheritance_denovo_only(inheritance_query: Sequence[str]) → bool[source]

Check if inheritance query is de novo only.

static check_inheritance_query_value(inheritance_query: Sequence[str], value: int) → bool[source]

Check if value satisfies a given inheritance query.

static check_roles_denovo_only(roles_query: str) → bool[source]

Check if roles query is de novo only.

static check_roles_query_value(roles_query: str, value: int) → bool[source]

Check if value satisfies a given roles query.

static check_sexes_query_value(sexes_query: str, value: int) → bool[source]

Check if value matches a given sexes query.

static check_variant_types_value(variant_types_query: str, value: int) → bool[source]

Check if value satisfies a given variant types query.

class dae.query_variants.sql.schema2.sql_query_builder.QueryHeuristics(region_bins: list[str], coding_bins: list[str], frequency_bins: list[str], family_bins: list[str])[source]

Bases: object

Heuristics for a query.

coding_bins: list[str]
family_bins: list[str]
frequency_bins: list[str]
is_empty() → bool[source]

Check if all heuristics are empty.

region_bins: list[str]
class dae.query_variants.sql.schema2.sql_query_builder.SqlQueryBuilder(db_layout: Db2Layout, *, schema: Schema, partition_descriptor: PartitionDescriptor | None, families: FamiliesData, gene_models: GeneModels, reference_genome: ReferenceGenome)[source]

Bases: QueryBuilderBase

Build SQL queries using sqlglot.

apply_family_heuristics(query: Select, heuristics: QueryHeuristics | None) → Select[source]

Apply heuristics to the family query.

apply_summary_heuristics(query: Select, heuristics: QueryHeuristics | None, table: str = 'sa') → Select[source]

Apply heuristics to the summary query.

static build(db_layout: Db2Layout, *, pedigree_schema: dict[str, str], summary_schema: dict[str, str], family_schema: dict[str, str], partition_descriptor: PartitionDescriptor | None, families: FamiliesData, gene_models: GeneModels, reference_genome: ReferenceGenome) → SqlQueryBuilder[source]

Return a new instance of the builder.

build_family_variants_query(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, family_ids: Sequence[str] | None = None, person_ids: Sequence[str] | None = None, inheritance: Sequence[str] | None = None, roles: str | None = None, sexes: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None, **_kwargs: Any) → list[str][source]

Build a query for family variants.

build_summary_variants_query(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None, limit: int | None = None) → list[str][source]

Build a query for summary variants.

static coding_bins(table: str, coding_bins: list[str]) → Condition[source]

Create coding bins condition.

static effect_types(effect_types: list[str]) → Condition[source]

Create effect types condition.

static family_base() → Select[source]
static family_bins(table: str, family_bins: list[str]) → Condition[source]

Create family bins condition.

static family_ids(family_ids: Sequence[str]) → Condition[source]

Create family IDs filter.

family_query(family_ids: Sequence[str] | None = None, person_ids: Sequence[str] | None = None, inheritance: str | Sequence[str] | None = None, roles: str | None = None, sexes: str | None = None) → Select[source]

Build a family subclause query.

family_variants(summary: Select, family: Select) → Select[source]

Construct family variants query.

static frequency(real_attrs: list[tuple[str, tuple[float | None, float | None]]]) → Condition[source]

Build frequencies filter where condition.

static frequency_bins(table: str, frequency_bins: list[str]) → Condition[source]

Create frequency bins condition.

static genes(genes: list[str]) → Condition[source]

Create genes condition.

static inheritance(inheritance_query: str | Sequence[str]) → Condition[source]

Build inheritance filter.

static person_ids(person_ids: Sequence[str]) → Condition[source]

Create person IDs filter.

static real_attr(real_attrs: list[tuple[str, tuple[float | None, float | None]]]) → Condition[source]

Build real attributes filter where condition.

static region_bins(table: str, region_bins: list[str]) → Condition[source]

Create region bins condition.

static regions(regions: list[Region]) → Condition[source]

Create regions condition.

replace_tables(query: Select) → Select[source]

Replace table names in the query.

static roles(roles_query: str) → Condition[source]
static sexes(sexes_query: str) → Condition[source]
static summary_base() → Select[source]

Create summary base query.

summary_query(*, regions: list[Region] | None = None, genes: list[str] | None = None, effect_types: list[str] | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, ultra_rare: bool | None = None, frequency_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, return_reference: bool | None = None, return_unknown: bool | None = None) → Select[source]

Build a summary variant query.

summary_variants(summary: Select) → Select[source]

Construct summary variants query.

static ultra_rare() → Condition[source]

dae.query_variants.sql.schema2.summary_builder module

class dae.query_variants.sql.schema2.summary_builder.SummaryQueryBuilder(dialect: Dialect, db: str | None, family_variant_table: str | None, summary_allele_table: str, pedigree_table: str, family_variant_schema: dict[str, str], summary_allele_schema: dict[str, str], table_properties: dict | None, pedigree_schema: dict[str, str], families: FamiliesData, gene_models: GeneModels | None = None)[source]

Bases: BaseQueryBuilder

Build queries related to summary variants.

Module contents