<s> import json <EOL> from simpleeval import InvalidExpression <EOL> from corehq . apps . userreports . document_stores import get_document_store <EOL> from corehq . apps . userreports . exceptions import BadSpecError <EOL> from corehq . util . couch import get_db_by_doc_type <EOL> from dimagi . ext . jsonobject import JsonObject , StringProperty , ListProperty , DictProperty <EOL> from jsonobject . base_properties import DefaultProperty <EOL> from corehq . apps . userreports . expressions . getters import ( <EOL> DictGetter , <EOL> NestedDictGetter , <EOL> TransformedGetter , <EOL> transform_from_datatype ) <EOL> from corehq . apps . userreports . indicators . specs import DataTypeProperty <EOL> from corehq . apps . userreports . specs import TypeProperty , EvaluationContext <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface <EOL> from pillowtop . dao . exceptions import DocumentNotFoundError <EOL> from . utils import eval_statements <EOL> from corehq . util . quickcache import quickcache <EOL> class IdentityExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> def __call__ ( self , item , context = None ) : <EOL> return item <EOL> class IterationNumberExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> def __call__ ( self , item , context = None ) : <EOL> return context . iteration <EOL> class ConstantGetterSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> constant = DefaultProperty ( ) <EOL> @ classmethod <EOL> def wrap ( self , obj ) : <EOL> if '<STR_LIT>' not in obj : <EOL> raise BadSpecError ( '<STR_LIT>' ) <EOL> return super ( ConstantGetterSpec , self ) . wrap ( obj ) <EOL> def __call__ ( self , item , context = None ) : <EOL> return self . constant <EOL> class PropertyNameGetterSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> property_name = StringProperty ( required = True ) <EOL> datatype = DataTypeProperty ( required = False ) <EOL> @ property <EOL> def expression ( self ) : <EOL> transform = transform_from_datatype ( self . datatype ) <EOL> getter = DictGetter ( self . property_name ) <EOL> return TransformedGetter ( getter , transform ) <EOL> def __call__ ( self , item , context = None ) : <EOL> return self . expression ( item , context ) <EOL> class PropertyPathGetterSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> property_path = ListProperty ( unicode , required = True ) <EOL> datatype = DataTypeProperty ( required = False ) <EOL> @ property <EOL> def expression ( self ) : <EOL> transform = transform_from_datatype ( self . datatype ) <EOL> getter = NestedDictGetter ( self . property_path ) <EOL> return TransformedGetter ( getter , transform ) <EOL> def __call__ ( self , item , context = None ) : <EOL> return self . expression ( item , context ) <EOL> class NamedExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> name = StringProperty ( required = True ) <EOL> def configure ( self , context ) : <EOL> if self . name not in context . named_expressions : <EOL> raise BadSpecError ( u'<STR_LIT>' . format ( self . name ) ) <EOL> self . _context = context <EOL> def __call__ ( self , item , context = None ) : <EOL> return self . _context . named_expressions [ self . name ] ( item , context ) <EOL> class ConditionalExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> test = DictProperty ( required = True ) <EOL> expression_if_true = DefaultProperty ( required = True ) <EOL> expression_if_false = DefaultProperty ( required = True ) <EOL> def configure ( self , test_function , true_expression , false_expression ) : <EOL> self . _test_function = test_function <EOL> self . _true_expression = true_expression <EOL> self . _false_expression = false_expression <EOL> def __call__ ( self , item , context = None ) : <EOL> if self . _test_function ( item , context ) : <EOL> return self . _true_expression ( item , context ) <EOL> else : <EOL> return self . _false_expression ( item , context ) <EOL> class ArrayIndexExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> array_expression = DictProperty ( required = True ) <EOL> index_expression = DefaultProperty ( required = True ) <EOL> def configure ( self , array_expression , index_expression ) : <EOL> self . _array_expression = array_expression <EOL> self . _index_expression = index_expression <EOL> def __call__ ( self , item , context = None ) : <EOL> array_value = self . _array_expression ( item , context ) <EOL> if not isinstance ( array_value , list ) : <EOL> return None <EOL> index_value = self . _index_expression ( item , context ) <EOL> if not isinstance ( index_value , int ) : <EOL> return None <EOL> try : <EOL> return array_value [ index_value ] <EOL> except IndexError : <EOL> return None <EOL> class SwitchExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> switch_on = DefaultProperty ( required = True ) <EOL> cases = DefaultProperty ( required = True ) <EOL> default = DefaultProperty ( required = True ) <EOL> def configure ( self , switch_on_expression , case_expressions , default_expression ) : <EOL> self . _switch_on_expression = switch_on_expression <EOL> self . _case_expressions = case_expressions <EOL> self . _default_expression = default_expression <EOL> def __call__ ( self , item , context = None ) : <EOL> switch_value = self . _switch_on_expression ( item , context ) <EOL> for c in self . cases : <EOL> if switch_value == c : <EOL> return self . _case_expressions [ c ] ( item , context ) <EOL> return self . _default_expression ( item , context ) <EOL> class IteratorExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> expressions = ListProperty ( required = True ) <EOL> test = DictProperty ( ) <EOL> def configure ( self , expressions , test ) : <EOL> self . _expression_fns = expressions <EOL> if test : <EOL> self . _test = test <EOL> else : <EOL> self . _test = lambda * args , ** kwargs : True <EOL> def __call__ ( self , item , context = None ) : <EOL> values = [ ] <EOL> for expression in self . _expression_fns : <EOL> value = expression ( item , context ) <EOL> if self . _test ( value ) : <EOL> values . append ( value ) <EOL> return values <EOL> class RootDocExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> expression = DictProperty ( required = True ) <EOL> def configure ( self , expression ) : <EOL> self . _expression_fn = expression <EOL> def __call__ ( self , item , context = None ) : <EOL> if context is None : <EOL> return None <EOL> return self . _expression_fn ( context . root_doc , context ) <EOL> class RelatedDocExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> related_doc_type = StringProperty ( ) <EOL> doc_id_expression = DictProperty ( required = True ) <EOL> value_expression = DictProperty ( required = True ) <EOL> def configure ( self , doc_id_expression , value_expression ) : <EOL> if get_db_by_doc_type ( self . related_doc_type ) is None : <EOL> raise BadSpecError ( u'<STR_LIT>' . format ( self . related_doc_type ) ) <EOL> self . _doc_id_expression = doc_id_expression <EOL> self . _value_expression = value_expression <EOL> self . _vary_on = json . dumps ( self . value_expression , sort_keys = True ) <EOL> def __call__ ( self , item , context = None ) : <EOL> doc_id = self . _doc_id_expression ( item , context ) <EOL> if doc_id : <EOL> return self . get_value ( doc_id , context ) <EOL> @ quickcache ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def get_value ( self , doc_id , context ) : <EOL> try : <EOL> assert context . root_doc [ '<STR_LIT>' ] <EOL> document_store = get_document_store ( context . root_doc [ '<STR_LIT>' ] , self . related_doc_type ) <EOL> doc = document_store . get_document ( doc_id ) <EOL> if context . root_doc [ '<STR_LIT>' ] != doc . get ( '<STR_LIT>' ) : <EOL> return None <EOL> return self . _value_expression ( doc , EvaluationContext ( doc , <NUM_LIT:0> ) ) <EOL> except DocumentNotFoundError : <EOL> return None <EOL> class NestedExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> argument_expression = DictProperty ( required = True ) <EOL> value_expression = DictProperty ( required = True ) <EOL> def configure ( self , argument_expression , value_expression ) : <EOL> self . _argument_expression = argument_expression <EOL> self . _value_expression = value_expression <EOL> def __call__ ( self , item , context = None ) : <EOL> argument = self . _argument_expression ( item , context ) <EOL> return self . _value_expression ( argument , context ) <EOL> class DictExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> properties = DictProperty ( required = True ) <EOL> def configure ( self , compiled_properties ) : <EOL> for key in compiled_properties : <EOL> if not isinstance ( key , basestring ) : <EOL> raise BadSpecError ( "<STR_LIT>" ) <EOL> self . _compiled_properties = compiled_properties <EOL> def __call__ ( self , item , context = None ) : <EOL> ret = { } <EOL> for property_name , expression in self . _compiled_properties . items ( ) : <EOL> ret [ property_name ] = expression ( item , context ) <EOL> return ret <EOL> class EvalExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> statement = StringProperty ( required = True ) <EOL> context_variables = DictProperty ( required = True ) <EOL> def configure ( self , context_variables ) : <EOL> self . _context_variables = context_variables <EOL> def __call__ ( self , item , context = None ) : <EOL> var_dict = self . get_variables ( item , context ) <EOL> try : <EOL> return eval_statements ( self . statement , var_dict ) <EOL> except ( InvalidExpression , SyntaxError ) : <EOL> return None <EOL> def get_variables ( self , item , context ) : <EOL> var_dict = { <EOL> slug : variable_expression ( item , context ) <EOL> for slug , variable_expression in self . _context_variables . items ( ) <EOL> } <EOL> return var_dict <EOL> class FormsExpressionSpec ( JsonObject ) : <EOL> type = TypeProperty ( '<STR_LIT>' ) <EOL> case_id_expression = DefaultProperty ( required = True ) <EOL> def configure ( self , case_id_expression ) : <EOL> self . _case_id_expression = case_id_expression <EOL> def __call__ ( self , item , context = None ) : <EOL> case_id = self . _case_id_expression ( item , context ) <EOL> if not case_id : <EOL> return [ ] <EOL> assert context . root_doc [ '<STR_LIT>' ] <EOL> return self . _get_forms ( case_id , context ) <EOL> def _get_forms ( self , case_id , context ) : <EOL> domain = context . root_doc [ '<STR_LIT>' ] <EOL> cache_key = ( self . __class__ . __name__ , case_id ) <EOL> if context . get_cache_value ( cache_key ) is not None : <EOL> return context . get_cache_value ( cache_key ) <EOL> xforms = FormProcessorInterface ( domain ) . get_case_forms ( case_id ) <EOL> xforms = [ f . to_json ( ) for f in xforms if f . domain == domain ] <EOL> context . set_cache_value ( cache_key , xforms ) <EOL> return xforms </s>
