Handle lag, issues & as_of

Author: rzats

src/server/_query.py

@@ -466,6 +466,27 @@ def where_time_pair(
         )
         return self
 
+    def apply_lag_filter(self, history_table: str, lag: Optional[int]):
+        if lag is not None:
+            self.retable(history_table)
+            # history_table has full spectrum of lag values to search from whereas the latest_table does not
+            self.where(lag=lag)
+
+    def apply_issues_filter(self, history_table: str, issues: Optional[TimeValues]):
+        if issues:
+            self.retable(history_table)
+            self.where_integers("issue", issues)
+
+    def apply_as_of_filter(self, history_table: str, as_of: Optional[int]):
+        if as_of is not None:
+            self.retable(history_table)
+            sub_condition_asof = "(issue <= :as_of)"
+            self.params["as_of"] = as_of
+            sub_fields = "max(issue) max_issue, time_type, time_value, `source`, `signal`, geo_type, geo_value"
+            sub_group = "time_type, time_value, `source`, `signal`, geo_type, geo_value"
+            sub_condition = f"x.max_issue = {self.alias}.issue AND x.time_type = {self.alias}.time_type AND x.time_value = {self.alias}.time_value AND x.source = {self.alias}.source AND x.signal = {self.alias}.signal AND x.geo_type = {self.alias}.geo_type AND x.geo_value = {self.alias}.geo_value"
+            self.subquery = f"JOIN (SELECT {sub_fields} FROM {self.table} WHERE {self.conditions_clause} AND {sub_condition_asof} GROUP BY {sub_group}) x ON {sub_condition}"
+
     def set_fields(self, *fields: Iterable[str]) -> "QueryBuilder":
         self.fields = [f"{self.alias}.{field}" for field_list in fields for field in field_list]
         return self

src/server/endpoints/covidcast.py

@@ -91,28 +91,6 @@ def parse_time_pairs() -> TimePair:
     return parse_time_arg()
 
 
-def _handle_lag_issues_as_of(q: QueryBuilder, issues: Optional[TimeValues] = None, lag: Optional[int] = None, as_of: Optional[int] = None):
-    if issues:
-        q.retable(history_table)
-        q.where_integers("issue", issues)
-    elif lag is not None:
-        q.retable(history_table)
-        # history_table has full spectrum of lag values to search from whereas the latest_table does not
-        q.where(lag=lag)
-    elif as_of is not None:
-        # fetch the most recent issue as of a certain date (not to be confused w/ plain-old "most recent issue"
-        q.retable(history_table)
-        sub_condition_asof = "(issue <= :as_of)"
-        q.params["as_of"] = as_of
-        sub_fields = "max(issue) max_issue, time_type, time_value, `source`, `signal`, geo_type, geo_value"
-        sub_group = "time_type, time_value, `source`, `signal`, geo_type, geo_value"
-        sub_condition = f"x.max_issue = {q.alias}.issue AND x.time_type = {q.alias}.time_type AND x.time_value = {q.alias}.time_value AND x.source = {q.alias}.source AND x.signal = {q.alias}.signal AND x.geo_type = {q.alias}.geo_type AND x.geo_value = {q.alias}.geo_value"
-        q.subquery = f"JOIN (SELECT {sub_fields} FROM {q.table} WHERE {q.conditions_clause} AND {sub_condition_asof} GROUP BY {sub_group}) x ON {sub_condition}"
-    else:
-        # else we are using the (standard/default) `latest_table`, to fetch the most recent issue quickly
-        pass
-
-
 @bp.route("/", methods=("GET", "POST"))
 def handle():
     source_signal_pairs = parse_source_signal_pairs()
@@ -147,7 +125,9 @@ def handle():
     q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
     q.where_time_pair("time_type", "time_value", time_pair)
 
-    _handle_lag_issues_as_of(q, issues, lag, as_of)
+    q.apply_issues_filter(history_table, issues)
+    q.apply_lag_filter(history_table, lag)
+    q.apply_as_of_filter(history_table, as_of)
 
     def transform_row(row, proxy):
         if is_compatibility or not alias_mapper or "source" not in row:
@@ -201,9 +181,6 @@ def handle_trend():
     q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
     q.where_time_pair("time_type", "time_value", time_window)
 
-    # fetch most recent issue fast
-    _handle_lag_issues_as_of(q, None, None, None)
-
     p = create_printer()
 
     def gen(rows):
@@ -252,9 +229,6 @@ def handle_trendseries():
     q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
     q.where_time_pair("time_type", "time_value", time_window)
 
-    # fetch most recent issue fast
-    _handle_lag_issues_as_of(q, None, None, None)
-
     p = create_printer()
 
     shifter = lambda x: shift_day_value(x, -basis_shift)
@@ -386,7 +360,7 @@ def handle_export():
     q.where_time_pair("time_type", "time_value", TimePair("day" if is_day else "week", [(start_day, end_day)]))
     q.where_geo_pairs("geo_type", "geo_value", [GeoPair(geo_type, True if geo_values == "*" else geo_values)])
 
-    _handle_lag_issues_as_of(q, None, None, as_of)
+    q.apply_as_of_filter(history_table, as_of)
 
     format_date = time_value_to_iso if is_day else lambda x: time_value_to_week(x).cdcformat()
     # tag as_of in filename, if it was specified
@@ -466,9 +440,6 @@ def handle_backfill():
     q.where_geo_pairs("geo_type", "geo_value", [geo_pair])
     q.where_time_pair("time_type", "time_value", time_pair)
 
-    # no restriction of issues or dates since we want all issues
-    # _handle_lag_issues_as_of(q, issues, lag, as_of)
-
     p = create_printer()
 
     def find_anchor_row(rows: List[Dict[str, Any]], issue: int) -> Optional[Dict[str, Any]]:
@@ -644,8 +615,6 @@ def handle_coverage():
     q.group_by = "c.source, c.signal, c.time_value"
     q.set_order("source", "signal", "time_value")
 
-    _handle_lag_issues_as_of(q, None, None, None)
-
     def transform_row(row, proxy):
         if not alias_mapper or "source" not in row:
             return row