import copy
from functools import partial
from fireant.dataset.fields import Field
from fireant.dataset.modifiers import Rollup
from pypika.terms import (
ComplexCriterion,
Criterion,
Term,
)
from .field_helper import make_term_for_dimension
[docs]def adapt_for_reference_query(reference_parts, database, dimensions, metrics, filters, references):
if reference_parts is None:
return dimensions, metrics, filters
ref_dimension, time_unit, interval = reference_parts
# Unpack rolled up dimensions
ref_dimension = ref_dimension.dimension if isinstance(ref_dimension, Rollup) else ref_dimension
ref_metrics = _make_reference_metrics(metrics,
references[0].reference_type.alias)
offset_func = partial(database.date_add,
date_part=time_unit,
interval=interval)
ref_dimensions = _make_reference_dimensions(database,
dimensions,
ref_dimension,
offset_func)
ref_filters = _make_reference_filters(filters,
ref_dimension,
offset_func)
return ref_dimensions, ref_metrics, ref_filters
def _make_reference_dimensions(database, dimensions, ref_dimension, offset_func):
def replace_reference_dimension(dimension):
ref_dimension_copy = copy.copy(dimension)
if hasattr(ref_dimension_copy, 'dimension'):
ref_dimension_copy.dimension = copy.copy(dimension.dimension)
ref_definition = make_term_for_dimension(ref_dimension_copy, database.trunc_date)
ref_dimension_copy.definition = offset_func(ref_definition)
return ref_dimension_copy
return [replace_reference_dimension(dimension)
if dimension is ref_dimension
else dimension
for dimension in dimensions]
def _make_reference_metrics(metrics, ref_key):
return [Field(metric.alias + '_' + ref_key,
metric.definition,
label=metric.label,
prefix=metric.prefix,
suffix=metric.suffix,
precision=metric.precision)
for metric in metrics]
def _make_reference_filters(filters, ref_dimension, offset_func):
"""
Copies and replaces the reference dimension's definition in all of the filters applied to a slicer query.
This is used to shift the dimension filters to fit the reference window.
:param filters:
:param ref_dimension:
:param offset_func:
:return:
"""
offset_ref_dimension_definition = offset_func(ref_dimension.definition)
reference_filters = []
for ref_filter in map(copy.deepcopy, filters):
ref_filter.definition = _apply_to_term_in_criterion(ref_dimension.definition,
offset_ref_dimension_definition,
ref_filter.definition)
reference_filters.append(ref_filter)
return reference_filters
def _apply_to_term_in_criterion(target: Term,
replacement: Term,
criterion: Criterion):
"""
Finds and replaces a term within a criterion. This is necessary for adapting filters used in reference queries
where the reference dimension must be offset by some value. The target term is found inside the criterion and
replaced with the replacement.
:param target:
The target term to replace in the criterion. It will be replaced in all locations within the criterion with
the func applied to itself.
:param replacement:
The replacement for the term.
:param criterion:
The criterion to replace the term in.
:return:
A criterion identical to the original criterion arg except with the target term replaced by the replacement arg.
"""
if isinstance(criterion, ComplexCriterion):
criterion.left = _apply_to_term_in_criterion(target, replacement, criterion.left)
criterion.right = _apply_to_term_in_criterion(target, replacement, criterion.right)
return criterion
for attr in ['term', 'left', 'right']:
if hasattr(criterion, attr) and str(getattr(criterion, attr)) == str(target):
setattr(criterion, attr, replacement)
return criterion