dae.query_variants.sql.schema2 package

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)

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, affected_statuses: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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) → str

Build an SQL query in the correct order.

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

Bases: object

Caries info about a SQL dialect.

static add_unnest_in_join() → bool

static array_item_suffix() → str

build_array_join(col: str, alias: str) → str

build_table_name(table: str, db: str | None) → str

static escape_char() → str

static escape_quote_char() → str

static float_type() → str

static int_type() → str

static string_type() → str

static use_bit_and_function() → bool

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)

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_in_parent: str | None = None, roles_in_child: str | None = None, roles: str | None = None, sexes: str | None = None, affected_statuses: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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, tags_query: TagsQuery | None = None, **kwargs: Any) → QueryRunner | None

Build a query selecting the appropriate family variants.

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, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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 | None

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]

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_in_parent: str | None = None, roles_in_child: str | None = None, sexes: str | None = None, affected_status: 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[FamilyVariant, None, None]

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_allele_in_members: bool = False)

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)

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)

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]

Return all region bins.

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

Build a list of regions based on genes.

static build_schema(summary_schema: dict[str, str], family_schema: dict[str, str], pedigree_schema: dict[str, str]) → Schema

static build_variant_types_query(variant_types_query: str) → Expression

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]

Calculate heuristics baches for a query.

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

Calculate applicable coding bins for a query.

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

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]

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

Calculate heuristic bins for a query.

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

Calculate applicable region bins for a query.

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

Check if inheritance query is de novo only.

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

Check if value satisfies a given inheritance query.

static check_roles_denovo_only(roles_query: str) → bool

Check if roles query is de novo only.

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

Check if value satisfies a given roles query.

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

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])

Bases: object

Heuristics for a query.

coding_bins: list[str]

family_bins: list[str]

frequency_bins: list[str]

is_empty() → bool

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)

Bases: QueryBuilderBase

Build SQL queries using sqlglot.

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

Apply heuristics to the family query.

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

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

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, affected_statuses: str | None = None, variant_type: str | None = None, real_attr_filter: list[tuple[str, tuple[float | None, float | None]]] | None = None, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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, tags_query: TagsQuery | None = None) → list[str]

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, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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]

Build a query for summary variants.

static categorical_attr(categorical_attrs: list[tuple[str, list[str] | list[int] | None]]) → Condition

Build real attributes filter where condition.

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

Create coding bins condition.

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

Create effect types condition.

static family_base() → Select

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

Create family bins condition.

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

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, affected_statuses: str | None = None, tags_query: TagsQuery | None = None) → Select

Build a family subclause query.

family_variants(summary: Select, family: Select) → Select

Construct family variants query.

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

Build frequencies filter where condition.

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

Create frequency bins condition.

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

Create genes condition.

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

Build inheritance filter.

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

Create person IDs filter.

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

Build real attributes filter where condition.

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

Create region bins condition.

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

Create regions condition.

replace_tables(query: Select) → Select

Replace table names in the query.

static resolve_tags(tags_query: TagsQuery, pedigree_table: Table) → Expression | None

Resolve tags query to an expression to use as a condition.

static roles(roles_query: str) → Expression

Construct a roles query condition.

Can match for zygosity in roles with a precalculated mask.

static sexes(sexes_query: str) → Expression

Construct a sexes query condition.

Can match for zygosity in sexes with a precalculated mask.

static statuses(statuses_query: str) → Expression

static summary_base() → Select

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, categorical_attr_filter: list[tuple[str, list[str] | list[int] | 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

Build a summary variant query.

summary_variants(summary: Select) → Select

Construct summary variants query.

static ultra_rare() → Condition

class dae.query_variants.sql.schema2.sql_query_builder.TagsQuery(*, selected_family_tags: list[str] | None = None, deselected_family_tags: list[str] | None = None, tags_or_mode: bool = False)

Bases: BaseModel

deselected_family_tags: list[str] | None

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'deselected_family_tags': FieldInfo(annotation=Union[list[str], NoneType], required=False, default=None), 'selected_family_tags': FieldInfo(annotation=Union[list[str], NoneType], required=False, default=None), 'tags_or_mode': FieldInfo(annotation=bool, required=False, default=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

selected_family_tags: list[str] | None

tags_or_mode: bool

class dae.query_variants.sql.schema2.sql_query_builder.ZygosityQuery(*, status_zygosity: str | None = None, parents_zygosity: int | None = None, children_zygosity: int | None = None, sex_zygosity: int | None = None)

Bases: BaseModel

children_zygosity: int | None

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'children_zygosity': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'parents_zygosity': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'sex_zygosity': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'status_zygosity': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

parents_zygosity: int | None

sex_zygosity: int | None

status_zygosity: str | None

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)

Bases: BaseQueryBuilder

Build queries related to summary variants.

Module contents