| 1096 | |
| 1097 | |
| 1098 | class AnalyzedFunction: |
| 1099 | __slots__ = ( |
| 1100 | "analyzed_code", |
| 1101 | "fn", |
| 1102 | "closure_pywrappers", |
| 1103 | "tracker_instrumented_fn", |
| 1104 | "expr", |
| 1105 | "bindparam_trackers", |
| 1106 | "expected_expr", |
| 1107 | "is_sequence", |
| 1108 | "propagate_attrs", |
| 1109 | "closure_bindparams", |
| 1110 | ) |
| 1111 | |
| 1112 | closure_bindparams: Optional[List[BindParameter[Any]]] |
| 1113 | expected_expr: Union[ClauseElement, List[ClauseElement]] |
| 1114 | bindparam_trackers: Optional[List[_BoundParameterGetter]] |
| 1115 | |
| 1116 | def __init__( |
| 1117 | self, |
| 1118 | analyzed_code, |
| 1119 | lambda_element, |
| 1120 | apply_propagate_attrs, |
| 1121 | fn, |
| 1122 | ): |
| 1123 | self.analyzed_code = analyzed_code |
| 1124 | self.fn = fn |
| 1125 | |
| 1126 | self.bindparam_trackers = analyzed_code.bindparam_trackers |
| 1127 | |
| 1128 | self._instrument_and_run_function(lambda_element) |
| 1129 | |
| 1130 | self._coerce_expression(lambda_element, apply_propagate_attrs) |
| 1131 | |
| 1132 | def _instrument_and_run_function(self, lambda_element): |
| 1133 | analyzed_code = self.analyzed_code |
| 1134 | |
| 1135 | fn = self.fn |
| 1136 | self.closure_pywrappers = closure_pywrappers = [] |
| 1137 | |
| 1138 | build_py_wrappers = analyzed_code.build_py_wrappers |
| 1139 | |
| 1140 | if not build_py_wrappers: |
| 1141 | self.tracker_instrumented_fn = tracker_instrumented_fn = fn |
| 1142 | self.expr = lambda_element._invoke_user_fn(tracker_instrumented_fn) |
| 1143 | else: |
| 1144 | track_closure_variables = analyzed_code.track_closure_variables |
| 1145 | closure = fn.__closure__ |
| 1146 | |
| 1147 | # will form the __closure__ of the function when we rebuild it |
| 1148 | if closure: |
| 1149 | new_closure = { |
| 1150 | fv: cell.cell_contents |
| 1151 | for fv, cell in zip(fn.__code__.co_freevars, closure) |
| 1152 | } |
| 1153 | else: |
| 1154 | new_closure = {} |
| 1155 |
no outgoing calls
no test coverage detected