delphi_interim_query_builder_test_env.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "delphi-interim-query-builder-test-env.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/melange396/delphi-epidata/blob/main/delphi_interim_query_builder_test_env.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "m4pQNStu8TCy",
        "outputId": "3d492a02-2fce-4639-ed2a-b10c505d5ac9"
      },
      "source": [
        "from datetime import date, datetime\n",
        "from typing import (\n",
        "    Any,\n",
        "    Callable,\n",
        "    Dict,\n",
        "    Iterable,\n",
        "    List,\n",
        "    Optional,\n",
        "    Sequence,\n",
        "    Tuple,\n",
        "    Union,\n",
        "    cast,\n",
        "    Mapping,\n",
        ")\n",
        "\n",
        "from sqlalchemy import text\n",
        "from sqlalchemy.engine import RowProxy\n",
        "\n",
        "from ._common import db, app\n",
        "from ._db import metadata\n",
        "from ._printer import create_printer, APrinter\n",
        "from ._exceptions import DatabaseErrorException\n",
        "from ._validate import DateRange, extract_strings\n",
        "from ._params import GeoPair, SourceSignalPair, TimePair\n",
        "\n",
        "\n",
        "def date_string(value: int) -> str:\n",
        "    # converts a date integer (YYYYMMDD) into a date string (YYYY-MM-DD)\n",
        "    # $value: the date as an 8-digit integer\n",
        "    year = int(value / 10000) % 10000\n",
        "    month = int(value / 100) % 100\n",
        "    day = value % 100\n",
        "    return \"{0:04d}-{1:02d}-{2:02d}\".format(year, month, day)\n",
        "\n",
        "\n",
        "def to_condition(\n",
        "    field: str,\n",
        "    value: Union[Tuple[str, str], str, Tuple[int, int], int],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "    formatter=lambda x: x,\n",
        ") -> str:\n",
        "    if isinstance(value, (list, tuple)):\n",
        "        params[param_key] = formatter(value[0])\n",
        "        params[f\"{param_key}_2\"] = formatter(value[1])\n",
        "        return f\"{field} BETWEEN :{param_key} AND :{param_key}_2\"\n",
        "\n",
        "    params[param_key] = formatter(value)\n",
        "    return f\"{field} = :{param_key}\"\n",
        "\n",
        "# TODO: does this work for conditioning on latest issue? who knows\n",
        "def to_indicator(\n",
        "    field: str,\n",
        "    value: Tuple[str,str],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "    formatter=lambda x: x,\n",
        ") -> str:\n",
        "    if isinstance(value, tuple):\n",
        "        # params[param_key] = formatter(value[0])\n",
        "        # params[f\"{param_key}_2\"] = formatter(value[1])\n",
        "        return f\"CASE WHEN {value[0]} = {value[1]} THEN 1 ELSE 0 END AS `is_latest_issue`\"\n",
        "\n",
        "    # params[param_key] = formatter(value)\n",
        "    # return f\"{field} = :{param_key}\"\n",
        "\n",
        "\n",
        "def filter_values(\n",
        "    field: str,\n",
        "    values: Optional[Sequence[Union[Tuple[str, str], str, Tuple[int, int], int]]],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "    formatter=lambda x: x,\n",
        "):\n",
        "    if not values:\n",
        "        return \"FALSE\"\n",
        "    # builds a SQL expression to filter strings (ex: locations)\n",
        "    #   $field: name of the field to filter\n",
        "    #   $values: array of values\n",
        "    conditions = [to_condition(field, v, f\"{param_key}_{i}\", params, formatter) for i, v in enumerate(values)]\n",
        "    return f\"({' OR '.join(conditions)})\"\n",
        "\n",
        "\n",
        "def filter_strings(\n",
        "    field: str,\n",
        "    values: Optional[Sequence[Union[Tuple[str, str], str]]],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "):\n",
        "    return filter_values(field, values, param_key, params)\n",
        "\n",
        "\n",
        "def filter_integers(\n",
        "    field: str,\n",
        "    values: Optional[Sequence[Union[Tuple[int, int], int]]],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "):\n",
        "    return filter_values(field, values, param_key, params)\n",
        "\n",
        "\n",
        "def filter_dates(\n",
        "    field: str,\n",
        "    values: Optional[Sequence[Union[Tuple[int, int], int]]],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        "):\n",
        "    return filter_values(field, values, param_key, params, date_string)\n",
        "\n",
        "\n",
        "def filter_fields(generator: Iterable[Dict[str, Any]]):\n",
        "    fields = extract_strings(\"fields\")\n",
        "    if not fields:\n",
        "        yield from generator\n",
        "    else:\n",
        "        exclude_fields = {f[1:] for f in fields if f.startswith(\"-\")}\n",
        "        include_fields = [f for f in fields if not f.startswith(\"-\") and f not in exclude_fields]\n",
        "\n",
        "        for row in generator:\n",
        "            filtered = dict()\n",
        "            if include_fields:\n",
        "                # positive list\n",
        "                for field in include_fields:\n",
        "                    if field in row:\n",
        "                        filtered[field] = row[field]\n",
        "            elif exclude_fields:\n",
        "                # negative list\n",
        "                for k, v in row.items():\n",
        "                    if k not in exclude_fields:\n",
        "                        filtered[k] = v\n",
        "            yield filtered\n",
        "\n",
        "\n",
        "def filter_geo_pairs(\n",
        "    type_field: str,\n",
        "    value_field: str,\n",
        "    values: Sequence[GeoPair],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        ") -> str:\n",
        "    \"\"\"\n",
        "    returns the SQL sub query to filter by the given geo pairs\n",
        "    \"\"\"\n",
        "\n",
        "    def filter_pair(pair: GeoPair, i) -> str:\n",
        "        type_param = f\"{param_key}_{i}t\"\n",
        "        params[type_param] = pair.geo_type\n",
        "        if isinstance(pair.geo_values, bool) and pair.geo_values:\n",
        "            return f\"{type_field} = :{type_param}\"\n",
        "        return f\"({type_field} = :{type_param} AND \\\n",
        "            {filter_strings(value_field, cast(Sequence[str], pair.geo_values), type_param, params)})\"\n",
        "\n",
        "    parts = [filter_pair(p, i) for i, p in enumerate(values)]\n",
        "\n",
        "    if not parts:\n",
        "        # something has to be selected\n",
        "        return \"FALSE\"\n",
        "\n",
        "    return f\"({' OR '.join(parts)})\"\n",
        "\n",
        "\n",
        "def filter_source_signal_pairs(\n",
        "    source_field: str,\n",
        "    signal_field: str,\n",
        "    values: Sequence[SourceSignalPair],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        ") -> str:\n",
        "    \"\"\"\n",
        "    returns the SQL sub query to filter by the given source signal pairs\n",
        "    \"\"\"\n",
        "\n",
        "    def filter_pair(pair: SourceSignalPair, i) -> str:\n",
        "        source_param = f\"{param_key}_{i}t\"\n",
        "        params[source_param] = pair.source\n",
        "        if isinstance(pair.signal, bool) and pair.signal:\n",
        "            return f\"{source_field} = :{source_param}\"\n",
        "        return f\"({source_field} = :{source_param} AND {filter_strings(signal_field, cast(Sequence[str], pair.signal), source_param, params)})\"\n",
        "\n",
        "    parts = [filter_pair(p, i) for i, p in enumerate(values)]\n",
        "\n",
        "    if not parts:\n",
        "        # something has to be selected\n",
        "        return \"FALSE\"\n",
        "\n",
        "    return f\"({' OR '.join(parts)})\"\n",
        "\n",
        "\n",
        "def filter_time_pairs(\n",
        "    type_field: str,\n",
        "    time_field: str,\n",
        "    values: Sequence[TimePair],\n",
        "    param_key: str,\n",
        "    params: Dict[str, Any],\n",
        ") -> str:\n",
        "    \"\"\"\n",
        "    returns the SQL sub query to filter by the given time pairs\n",
        "    \"\"\"\n",
        "\n",
        "    def filter_pair(pair: TimePair, i) -> str:\n",
        "        type_param = f\"{param_key}_{i}t\"\n",
        "        params[type_param] = pair.time_type\n",
        "        if isinstance(pair.time_values, bool) and pair.time_values:\n",
        "            return f\"{type_field} = :{type_param}\"\n",
        "        return f\"({type_field} = :{type_param} AND\\\n",
        "             {filter_integers(time_field, cast(Sequence[Union[int, Tuple[int,int]]], pair.time_values), type_param, params)})\"\n",
        "\n",
        "    parts = [filter_pair(p, i) for i, p in enumerate(values)]\n",
        "\n",
        "    if not parts:\n",
        "        # something has to be selected\n",
        "        return \"FALSE\"\n",
        "\n",
        "    return f\"({' OR '.join(parts)})\"\n",
        "\n",
        "\n",
        "def parse_row(\n",
        "    row: RowProxy,\n",
        "    fields_string: Optional[Sequence[str]] = None,\n",
        "    fields_int: Optional[Sequence[str]] = None,\n",
        "    fields_float: Optional[Sequence[str]] = None,\n",
        "):\n",
        "    keys = set(row.keys())\n",
        "    parsed = dict()\n",
        "    if fields_string:\n",
        "        for f in fields_string:\n",
        "            v = row[f] if f in keys else None\n",
        "            if isinstance(v, (date, datetime)):\n",
        "                v = v.strftime(\"%Y-%m-%d\")  # format to iso date\n",
        "            parsed[f] = v\n",
        "    if fields_int:\n",
        "        for f in fields_int:\n",
        "            parsed[f] = int(row[f]) if f in keys and row[f] is not None else None\n",
        "    if fields_float:\n",
        "        for f in fields_float:\n",
        "            parsed[f] = float(row[f]) if f in keys and row[f] is not None else None\n",
        "    return parsed\n",
        "\n",
        "\n",
        "def parse_result(\n",
        "    query: str,\n",
        "    params: Dict[str, Any],\n",
        "    fields_string: Optional[Sequence[str]] = None,\n",
        "    fields_int: Optional[Sequence[str]] = None,\n",
        "    fields_float: Optional[Sequence[str]] = None,\n",
        ") -> List[Dict[str, Any]]:\n",
        "    \"\"\"\n",
        "    execute the given query and return the result as a list of dictionaries\n",
        "    \"\"\"\n",
        "    return [parse_row(row, fields_string, fields_int, fields_float) for row in db.execute(text(query), **params)]\n",
        "\n",
        "\n",
        "def run_query(p: APrinter, query_tuple: Tuple[str, Dict[str, Any]]):\n",
        "    query, params = query_tuple\n",
        "    # limit rows + 1 for detecting whether we would have more\n",
        "    full_query = text(f\"{query} LIMIT {p.remaining_rows + 1}\")\n",
        "    app.logger.info(\"full_query: %s, params: %s\", full_query, params)\n",
        "    return db.execution_options(stream_results=True).execute(full_query, **params)\n",
        "\n",
        "\n",
        "def _identity_transform(row: Dict[str, Any], _: RowProxy) -> Dict[str, Any]:\n",
        "    \"\"\"\n",
        "    identity transform\n",
        "    \"\"\"\n",
        "    return row\n",
        "\n",
        "\n",
        "def execute_queries(\n",
        "    queries: Sequence[Tuple[str, Dict[str, Any]]],\n",
        "    fields_string: Sequence[str],\n",
        "    fields_int: Sequence[str],\n",
        "    fields_float: Sequence[str],\n",
        "    transform: Callable[[Dict[str, Any], RowProxy], Dict[str, Any]] = _identity_transform,\n",
        "):\n",
        "    \"\"\"\n",
        "    execute the given queries and return the response to send them\n",
        "    \"\"\"\n",
        "\n",
        "    p = create_printer()\n",
        "\n",
        "    fields_to_send = set(extract_strings(\"fields\") or [])\n",
        "    if fields_to_send:\n",
        "        exclude_fields = {f[1:] for f in fields_to_send if f.startswith(\"-\")}\n",
        "        include_fields = {f for f in fields_to_send if not f.startswith(\"-\") and f not in exclude_fields}\n",
        "\n",
        "        if include_fields:\n",
        "            fields_string = [v for v in fields_string if v in include_fields]\n",
        "            fields_int = [v for v in fields_int if v in include_fields]\n",
        "            fields_float = [v for v in fields_float if v in include_fields]\n",
        "        if exclude_fields:\n",
        "            fields_string = [v for v in fields_string if v not in exclude_fields]\n",
        "            fields_int = [v for v in fields_int if v not in exclude_fields]\n",
        "            fields_float = [v for v in fields_float if v not in exclude_fields]\n",
        "\n",
        "    query_list = list(queries)\n",
        "\n",
        "    def dummy_gen():\n",
        "        if 3 > 4:\n",
        "            yield {}\n",
        "        pass\n",
        "\n",
        "    if not query_list or p.remaining_rows <= 0:\n",
        "        return p(dummy_gen)\n",
        "\n",
        "    def gen(first_rows):\n",
        "        for row in first_rows:\n",
        "            yield transform(parse_row(row, fields_string, fields_int, fields_float), row)\n",
        "\n",
        "        for query_params in query_list:\n",
        "            if p.remaining_rows <= 0:\n",
        "                # no more rows\n",
        "                break\n",
        "            r = run_query(p, query_params)\n",
        "            for row in r:\n",
        "                yield transform(parse_row(row, fields_string, fields_int, fields_float), row)\n",
        "\n",
        "    # execute first query\n",
        "    try:\n",
        "        r = run_query(p, query_list.pop(0))\n",
        "    except Exception as e:\n",
        "        raise DatabaseErrorException(str(e))\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(gen(r))\n",
        "\n",
        "\n",
        "def execute_query(\n",
        "    query: str,\n",
        "    params: Dict[str, Any],\n",
        "    fields_string: Sequence[str],\n",
        "    fields_int: Sequence[str],\n",
        "    fields_float: Sequence[str],\n",
        "    transform: Callable[[Dict[str, Any], RowProxy], Dict[str, Any]] = _identity_transform,\n",
        "):\n",
        "    \"\"\"\n",
        "    execute the given query and return the response to send it\n",
        "    \"\"\"\n",
        "    return execute_queries([(query, params)], fields_string, fields_int, fields_float, transform)\n",
        "\n",
        "\n",
        "def _join_l(value: Union[str, List[str]]):\n",
        "    return \", \".join(value) if isinstance(value, (list, tuple)) else value\n",
        "\n",
        "\n",
        "class QueryBuilder:\n",
        "    \"\"\"\n",
        "    query builder helper class for simplified conditions\n",
        "    \"\"\"\n",
        "\n",
        "    def __init__(self, reftable:Union[str,\"QueryBuilder\"], refalias: str, datatable: str, dataalias: str):\n",
        "        self.reftable: Union[str,\"QueryBuilder\"] = \"\"\n",
        "        self.refalias: str = refalias\n",
        "        self.datatable: str = \"\"\n",
        "        self.dataalia: str = \"\"\n",
        "        self.group_by: Union[str, List[str]] = \"\"\n",
        "        self.order: Union[str, List[str]] = \"\"\n",
        "        self.fields: Union[str, List[str]] = \"*\"\n",
        "        self.conditions: List[str] = []\n",
        "        self.params: Dict[str, Any] = {}\n",
        "        self.subquery: str = \"\"\n",
        "        self.index: Optional[str] = None\n",
        "\n",
        "    @property\n",
        "    def conditions_clause(self) -> str:\n",
        "        return \" AND \".join(self.conditions)\n",
        "\n",
        "    @property\n",
        "    def fields_clause(self) -> str:\n",
        "        return _join_l(self.fields) if self.fields else \"*\"\n",
        "\n",
        "    @property\n",
        "    def order_clause(self) -> str:\n",
        "        return _join_l(self.order)\n",
        "\n",
        "    def __str__(self):\n",
        "        table = self.table if isinstance(self.table,str) else f\"({str(self.table)})\"\n",
        "        where = f\"WHERE {self.conditions_clause}\" if self.conditions else \"\"\n",
        "        order = f\"ORDER BY {_join_l(self.order)}\" if self.order else \"\"\n",
        "        group_by = f\"GROUP BY {_join_l(self.group_by)}\" if self.group_by else \"\"\n",
        "        index = f\"USE INDEX ({self.index})\" if self.index else \"\"\n",
        "\n",
        "        return f\"SELECT {self.fields_clause} FROM {self.table} {index} {self.subquery} {where} {group_by} {order}\"\n",
        "\n",
        "    @property\n",
        "    def query(self) -> str:\n",
        "        \"\"\"\n",
        "        returns the full query\n",
        "        \"\"\"\n",
        "        return str(self)\n",
        "\n",
        "    def where(self, **kvargs: Dict[str, Any]) -> \"QueryBuilder\":\n",
        "        for k, v in kvargs.items():\n",
        "            self.conditions.append(f\"{self.alias}.{k} = :{k}\")\n",
        "            self.params[k] = v\n",
        "        return self\n",
        "\n",
        "    def where_strings(\n",
        "        self,\n",
        "        field: str,\n",
        "        values: Optional[Sequence[Union[Tuple[str, str], str]]],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_field = f\"{self.alias}.{field}\" if \".\" not in field else field\n",
        "        self.conditions.append(filter_strings(fq_field, values, param_key or field, self.params))\n",
        "        return self\n",
        "\n",
        "    def _fq_field(self, field: str) -> str:\n",
        "        return f\"{self.alias}.{field}\" if \".\" not in field else field\n",
        "\n",
        "    def where_integers(\n",
        "        self,\n",
        "        field: str,\n",
        "        values: Optional[Sequence[Union[Tuple[int, int], int]]],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_field = self._fq_field(field)\n",
        "        self.conditions.append(filter_integers(fq_field, values, param_key or field, self.params))\n",
        "        return self\n",
        "\n",
        "    def where_dates(\n",
        "        self,\n",
        "        field: str,\n",
        "        values: Optional[Sequence[Union[Tuple[int, int], int]]],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_field = self._fq_field(field)\n",
        "        self.conditions.append(filter_dates(fq_field, values, param_key or field, self.params))\n",
        "        return self\n",
        "\n",
        "    def where_geo_pairs(\n",
        "        self,\n",
        "        type_field: str,\n",
        "        value_field: str,\n",
        "        values: Sequence[GeoPair],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_type_field = self._fq_field(type_field)\n",
        "        fq_value_field = self._fq_field(value_field)\n",
        "        self.conditions.append(\n",
        "            filter_geo_pairs(\n",
        "                fq_type_field,\n",
        "                fq_value_field,\n",
        "                values,\n",
        "                param_key or type_field,\n",
        "                self.params,\n",
        "            )\n",
        "        )\n",
        "        return self\n",
        "\n",
        "    def where_source_signal_pairs(\n",
        "        self,\n",
        "        type_field: str,\n",
        "        value_field: str,\n",
        "        values: Sequence[SourceSignalPair],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_type_field = self._fq_field(type_field)\n",
        "        fq_value_field = self._fq_field(value_field)\n",
        "        self.conditions.append(\n",
        "            filter_source_signal_pairs(\n",
        "                fq_type_field,\n",
        "                fq_value_field,\n",
        "                values,\n",
        "                param_key or type_field,\n",
        "                self.params,\n",
        "            )\n",
        "        )\n",
        "        return self\n",
        "\n",
        "    def where_time_pairs(\n",
        "        self,\n",
        "        type_field: str,\n",
        "        value_field: str,\n",
        "        values: Sequence[TimePair],\n",
        "        param_key: Optional[str] = None,\n",
        "    ) -> \"QueryBuilder\":\n",
        "        fq_type_field = self._fq_field(type_field)\n",
        "        fq_value_field = self._fq_field(value_field)\n",
        "        self.conditions.append(\n",
        "            filter_time_pairs(\n",
        "                fq_type_field,\n",
        "                fq_value_field,\n",
        "                values,\n",
        "                param_key or type_field,\n",
        "                self.params,\n",
        "            )\n",
        "        )\n",
        "        return self\n",
        "\n",
        "    def set_fields(self, *fields: Iterable[str]) -> \"QueryBuilder\":\n",
        "        self.fields = [f\"{self.alias}.{field}\" for field_list in fields for field in field_list]\n",
        "        return self\n",
        "\n",
        "    def set_order(self, *args: str, **kwargs: Union[str, bool]) -> \"QueryBuilder\":\n",
        "        \"\"\"\n",
        "        sets the order for the given fields (as key word arguments), True = ASC, False = DESC\n",
        "        \"\"\"\n",
        "\n",
        "        def to_asc(v: Union[str, bool]) -> str:\n",
        "            if v == True:\n",
        "                return \"ASC\"\n",
        "            elif v == False:\n",
        "                return \"DESC\"\n",
        "            return cast(str, v)\n",
        "\n",
        "        args_order = [f\"{self.alias}.{k} ASC\" for k in args]\n",
        "        kw_order = [f\"{self.alias}.{k} {to_asc(v)}\" for k, v in kwargs.items()]\n",
        "        self.order = args_order + kw_order\n",
        "        return self\n",
        "\n",
        "    def with_max_issue(self, *args: str) -> \"QueryBuilder\":\n",
        "        fields: List[str] = [f for f in args]\n",
        "\n",
        "        subfields = f\"max(issue) max_issue, {','.join(fields)}\"\n",
        "        group_by = \",\".join(fields)\n",
        "        field_conditions = \" AND \".join(f\"x.{field} = {self.alias}.{field}\" for field in fields)\n",
        "        condition = f\"x.max_issue = {self.alias}.issue AND {field_conditions}\"\n",
        "        self.subquery = f\"JOIN (SELECT {subfields} FROM {self.table} WHERE {self.conditions_clause} GROUP BY {group_by}) x ON {condition}\"\n",
        "        # reset conditions since for join\n",
        "        self.conditions = []\n",
        "        return self\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[1 2 3]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 368
        },
        "id": "zDSPke_j9dS0",
        "outputId": "94a3e711-b8fc-41f4-e032-3482b4ede834"
      },
      "source": [
        "from typing import List, Optional, Union, Tuple, Dict, Any, Set\n",
        "from itertools import groupby\n",
        "from datetime import date, datetime, timedelta\n",
        "from flask import Blueprint, request\n",
        "from flask.json import loads, jsonify\n",
        "from bisect import bisect_right\n",
        "from sqlalchemy import text\n",
        "from pandas import read_csv\n",
        "\n",
        "from .._common import is_compatibility_mode, db\n",
        "from .._exceptions import ValidationFailedException, DatabaseErrorException\n",
        "from .._params import (\n",
        "    GeoPair,\n",
        "    SourceSignalPair,\n",
        "    TimePair,\n",
        "    parse_geo_arg,\n",
        "    parse_source_signal_arg,\n",
        "    parse_time_arg,\n",
        "    parse_day_arg,\n",
        "    parse_day_range_arg,\n",
        "    parse_single_source_signal_arg,\n",
        "    parse_single_time_arg,\n",
        "    parse_single_geo_arg,\n",
        ")\n",
        "from .._query import QueryBuilder, execute_query, run_query, parse_row, filter_fields\n",
        "from .._printer import create_printer, CSVPrinter\n",
        "from .._validate import (\n",
        "    extract_date,\n",
        "    extract_dates,\n",
        "    extract_integer,\n",
        "    extract_strings,\n",
        "    require_all,\n",
        "    require_any,\n",
        ")\n",
        "from .._pandas import as_pandas, print_pandas\n",
        "from .covidcast_utils import compute_trend, compute_trends, compute_correlations, compute_trend_value, CovidcastMetaEntry, AllSignalsMap\n",
        "from ..utils import shift_time_value, date_to_time_value, time_value_to_iso, time_value_to_date\n",
        "\n",
        "# first argument is the endpoint name\n",
        "bp = Blueprint(\"covidcast\", __name__)\n",
        "alias = None\n",
        "\n",
        "\n",
        "def parse_source_signal_pairs() -> List[SourceSignalPair]:\n",
        "    ds = request.values.get(\"data_source\")\n",
        "    if ds:\n",
        "        # old version\n",
        "        require_any(\"signal\", \"signals\")\n",
        "        signals = extract_strings((\"signals\", \"signal\"))\n",
        "        return [SourceSignalPair(ds, signals)]\n",
        "\n",
        "    if \":\" not in request.values.get(\"signal\", \"\"):\n",
        "        raise ValidationFailedException(\"missing parameter: signal or (data_source and signal[s])\")\n",
        "\n",
        "    return parse_source_signal_arg()\n",
        "\n",
        "\n",
        "def parse_geo_pairs() -> List[GeoPair]:\n",
        "    geo_type = request.values.get(\"geo_type\")\n",
        "    if geo_type:\n",
        "        # old version\n",
        "        require_any(\"geo_value\", \"geo_values\", empty=True)\n",
        "        geo_values = extract_strings((\"geo_values\", \"geo_value\"))\n",
        "        if len(geo_values) == 1 and geo_values[0] == \"*\":\n",
        "            return [GeoPair(geo_type, True)]\n",
        "        return [GeoPair(geo_type, geo_values)]\n",
        "\n",
        "    if \":\" not in request.values.get(\"geo\", \"\"):\n",
        "        raise ValidationFailedException(\"missing parameter: geo or (geo_type and geo_value[s])\")\n",
        "\n",
        "    return parse_geo_arg()\n",
        "\n",
        "\n",
        "def parse_time_pairs() -> List[TimePair]:\n",
        "    time_type = request.values.get(\"time_type\")\n",
        "    if time_type:\n",
        "        # old version\n",
        "        require_all(\"time_type\", \"time_values\")\n",
        "        time_values = extract_dates(\"time_values\")\n",
        "        return [TimePair(time_type, time_values)]\n",
        "\n",
        "    if \":\" not in request.values.get(\"time\", \"\"):\n",
        "        raise ValidationFailedException(\"missing parameter: time or (time_type and time_values)\")\n",
        "\n",
        "    return parse_time_arg()\n",
        "\n",
        "\n",
        "def _handle_lag_issues_as_of(q: QueryBuilder, issues: Optional[List[Union[Tuple[int, int], int]]] = None, lag: Optional[int] = None, as_of: Optional[int] = None):\n",
        "    if issues:\n",
        "        q.where_integers(\"issue\", issues)\n",
        "    elif lag is not None:\n",
        "        q.where(lag=lag)\n",
        "    elif as_of is not None:\n",
        "        # fetch most recent issues with as of\n",
        "        sub_condition_asof = \"(issue <= :as_of)\"\n",
        "        q.params[\"as_of\"] = as_of\n",
        "        sub_fields = \"max(issue) max_issue, time_type, time_value, `source`, `signal`, geo_type, geo_value\"\n",
        "        sub_group = \"time_type, time_value, `source`, `signal`, geo_type, geo_value\"\n",
        "        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\"\n",
        "        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}\"\n",
        "    else:\n",
        "        # fetch most recent issue fast\n",
        "        q.conditions.append(f\"({q.alias}.is_latest_issue IS TRUE)\")\n",
        "\n",
        "\n",
        "def guess_index_to_use(time: List[TimePair], geo: List[GeoPair], issues: Optional[List[Union[Tuple[int, int], int]]] = None, lag: Optional[int] = None, as_of: Optional[int] = None) -> Optional[str]:\n",
        "    time_values_to_retrieve = sum((t.count() for t in time))\n",
        "    geo_values_to_retrieve = sum((g.count() for g in geo))\n",
        "\n",
        "    if geo_values_to_retrieve > 5 or time_values_to_retrieve < 30:\n",
        "        # no optimization known\n",
        "        return None\n",
        "\n",
        "    if issues:\n",
        "        return \"by_issue\"\n",
        "    elif lag is not None:\n",
        "        return \"by_lag\"\n",
        "    elif as_of is None:\n",
        "        # latest\n",
        "        return \"by_issue\"\n",
        "    return None\n",
        "\n",
        "\n",
        "@bp.route(\"/\", methods=(\"GET\", \"POST\"))\n",
        "def handle():\n",
        "    source_signal_pairs = parse_source_signal_pairs()\n",
        "    time_pairs = parse_time_pairs()\n",
        "    geo_pairs = parse_geo_pairs()\n",
        "\n",
        "    as_of = extract_date(\"as_of\")\n",
        "    issues = extract_dates(\"issues\")\n",
        "    lag = extract_integer(\"lag\")\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"data_reference\", \"ref\", \"datapoint\", \"point\")\n",
        "\n",
        "    fields_string = [\"ref.geo_value\", \"ref.signal\"]\n",
        "    fields_int = [\"ref.time_value\", \"point.issue\", \"point.lag\", \"point.missing_value\", \"point.missing_stderr\", \"point.missing_sample_size\"]\n",
        "    fields_float = [\"point.value\", \"point.stderr\", \"point.sample_size\"]\n",
        "\n",
        "    if is_compatibility_mode():\n",
        "        q.set_order(\"ref.signal\", \"ref.time_value\", \"ref.geo_value\", \"point.issue\")\n",
        "    else:\n",
        "        # transfer also the new detail columns\n",
        "        fields_string.extend([\"ref.source\", \"ref.geo_type\", \"ref.time_type\"])\n",
        "        q.set_order(\"ref.source\", \"ref.signal\", \"ref.time_type\", \"ref.time_value\", \"ref.geo_type\", \"ref.geo_value\", \"point.issue\")\n",
        "    q.set_fields(fields_string, fields_int, fields_float)\n",
        "\n",
        "    # basic query info\n",
        "    # data type of each field\n",
        "    # build the source, signal, time, and location (type and id) filters\n",
        "\n",
        "    q.where_source_signal_pairs(\"ref.source\", \"ref.signal\", source_signal_pairs)\n",
        "    q.where_geo_pairs(\"ref.geo_type\", \"ref.geo_value\", geo_pairs)\n",
        "    q.where_time_pairs(\"ref.time_type\", \"ref.time_value\", time_pairs)\n",
        "\n",
        "    # TODO: Reevaluate the guess_index_to_use function since it tends to choose incorrectly\n",
        "    # q.index = guess_index_to_use(time_pairs, geo_pairs, issues, lag, as_of)\n",
        "\n",
        "    _handle_lag_issues_as_of(q, issues, lag, as_of)\n",
        "\n",
        "    # send query\n",
        "    # TODO: establish a boolean for indicating is_latest_query vs as_of\n",
        "    is_latest_query = True\n",
        "    if is_latest_query:\n",
        "        return execute_query(str(q), q.params, fields_string, fields_int, fields_float)\n",
        "    else:\n",
        "        q = f'''SELECT * FROM( {str(q)}) all_issues\n",
        "        WHERE all_issues.issue <= all_issues.max_issue\n",
        "        ORDER BY source ASC, `signal` ASC, time_type ASC, time_value ASC, geo_type ASC, \n",
        "        geo_value ASC, issue ASC'''\n",
        "        return execute_query(str(q), q.params, fields_string, fields_int, fields_float)\n",
        "        # TODO: Calls query builder again around the returned values giving it alias all_issues\n",
        "       \n",
        "        # q = QueryBuilder(all_issues, \"all_issues\", \"data_reference\", \"ref\")\n",
        "        \n",
        "\n",
        "\n",
        "@bp.route(\"/trend\", methods=(\"GET\", \"POST\"))\n",
        "def handle_trend():\n",
        "    require_all(\"date\", \"window\")\n",
        "    source_signal_pairs = parse_source_signal_pairs()\n",
        "    geo_pairs = parse_geo_pairs()\n",
        "\n",
        "    time_value = parse_day_arg(\"date\")\n",
        "    time_window = parse_day_range_arg(\"window\")\n",
        "    basis_time_value = extract_date(\"basis\") or shift_time_value(time_value, -7)\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"datapoint\", \"t\")\n",
        "\n",
        "    fields_string = [\"geo_type\", \"geo_value\", \"source\", \"signal\"]\n",
        "    fields_int = [\"time_value\"]\n",
        "    fields_float = [\"value\"]\n",
        "    q.set_fields(fields_string, fields_int, fields_float)\n",
        "    q.set_order(\"geo_type\", \"geo_value\", \"source\", \"signal\", \"time_value\")\n",
        "\n",
        "    q.where_source_signal_pairs(\"source\", \"signal\", source_signal_pairs)\n",
        "    q.where_geo_pairs(\"geo_type\", \"geo_value\", geo_pairs)\n",
        "    q.where_time_pairs(\"time_type\", \"time_value\", [TimePair(\"day\", [time_window])])\n",
        "\n",
        "    # fetch most recent issue fast\n",
        "    _handle_lag_issues_as_of(q, None, None, None)\n",
        "\n",
        "    p = create_printer()\n",
        "\n",
        "    def gen(rows):\n",
        "        for key, group in groupby((parse_row(row, fields_string, fields_int, fields_float) for row in rows), lambda row: (row[\"geo_type\"], row[\"geo_value\"], row[\"source\"], row[\"signal\"])):\n",
        "            trend = compute_trend(key[0], key[1], key[2], key[3], time_value, basis_time_value, ((row[\"time_value\"], row[\"value\"]) for row in group))\n",
        "            yield trend.asdict()\n",
        "\n",
        "    # execute first query\n",
        "    try:\n",
        "        r = run_query(p, (str(q), q.params))\n",
        "    except Exception as e:\n",
        "        raise DatabaseErrorException(str(e))\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(filter_fields(gen(r)))\n",
        "\n",
        "\n",
        "@bp.route(\"/trendseries\", methods=(\"GET\", \"POST\"))\n",
        "def handle_trendseries():\n",
        "    require_all(\"window\")\n",
        "    source_signal_pairs = parse_source_signal_pairs()\n",
        "    geo_pairs = parse_geo_pairs()\n",
        "\n",
        "    time_window = parse_day_range_arg(\"window\")\n",
        "    basis_shift = extract_integer(\"basis\")\n",
        "    if basis_shift is None:\n",
        "        basis_shift = 7\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"datapoint\", \"t\")\n",
        "\n",
        "    fields_string = [\"geo_type\", \"geo_value\", \"source\", \"signal\"]\n",
        "    fields_int = [\"time_value\"]\n",
        "    fields_float = [\"value\"]\n",
        "    q.set_fields(fields_string, fields_int, fields_float)\n",
        "    q.set_order(\"geo_type\", \"geo_value\", \"source\", \"signal\", \"time_value\")\n",
        "\n",
        "    q.where_source_signal_pairs(\"source\", \"signal\", source_signal_pairs)\n",
        "    q.where_geo_pairs(\"geo_type\", \"geo_value\", geo_pairs)\n",
        "    q.where_time_pairs(\"time_type\", \"time_value\", [TimePair(\"day\", [time_window])])\n",
        "\n",
        "    # fetch most recent issue fast\n",
        "    _handle_lag_issues_as_of(q, None, None, None)\n",
        "\n",
        "    p = create_printer()\n",
        "\n",
        "    shifter = lambda x: shift_time_value(x, -basis_shift)\n",
        "\n",
        "    def gen(rows):\n",
        "        for key, group in groupby((parse_row(row, fields_string, fields_int, fields_float) for row in rows), lambda row: (row[\"geo_type\"], row[\"geo_value\"], row[\"source\"], row[\"signal\"])):\n",
        "            trends = compute_trends(key[0], key[1], key[2], key[3], shifter, ((row[\"time_value\"], row[\"value\"]) for row in group))\n",
        "            for trend in trends:\n",
        "                yield trend.asdict()\n",
        "\n",
        "    # execute first query\n",
        "    try:\n",
        "        r = run_query(p, (str(q), q.params))\n",
        "    except Exception as e:\n",
        "        raise DatabaseErrorException(str(e))\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(filter_fields(gen(r)))\n",
        "\n",
        "\n",
        "@bp.route(\"/correlation\", methods=(\"GET\", \"POST\"))\n",
        "def handle_correlation():\n",
        "    require_all(\"reference\", \"window\", \"others\", \"geo\")\n",
        "    reference = parse_single_source_signal_arg(\"reference\")\n",
        "    other_pairs = parse_source_signal_arg(\"others\")\n",
        "    geo_pairs = parse_geo_arg()\n",
        "    time_window = parse_day_range_arg(\"window\")\n",
        "    lag = extract_integer(\"lag\")\n",
        "    if lag is None:\n",
        "        lag = 28\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"datapoint\", \"t\")\n",
        "\n",
        "    fields_string = [\"geo_type\", \"geo_value\", \"source\", \"signal\"]\n",
        "    fields_int = [\"time_value\"]\n",
        "    fields_float = [\"value\"]\n",
        "    q.set_fields(fields_string, fields_int, fields_float)\n",
        "    q.set_order(\"geo_type\", \"geo_value\", \"source\", \"signal\", \"time_value\")\n",
        "\n",
        "    q.where_source_signal_pairs(\"source\", \"signal\", other_pairs + [reference])\n",
        "    q.where_geo_pairs(\"geo_type\", \"geo_value\", geo_pairs)\n",
        "    q.where_time_pairs(\"time_type\", \"time_value\", [TimePair(\"day\", [time_window])])\n",
        "\n",
        "    # fetch most recent issue fast\n",
        "    q.conditions.append(f\"({q.alias}.is_latest_issue IS TRUE)\")\n",
        "\n",
        "    df = as_pandas(str(q), q.params, parse_dates={\"time_value\": \"%Y%m%d\"})\n",
        "\n",
        "    p = create_printer()\n",
        "\n",
        "    def prepare_data_frame(df):\n",
        "        return df[[\"time_value\", \"value\"]].set_index(\"time_value\")\n",
        "\n",
        "    def gen():\n",
        "        by_geo = df.groupby([\"geo_type\", \"geo_value\"])\n",
        "        for (geo_type, geo_value), group in by_geo:\n",
        "            # group by source, signal\n",
        "            by_signal = group.groupby([\"source\", \"signal\"])\n",
        "\n",
        "            # find reference group\n",
        "            # dataframe structure: index=time_value, value=value\n",
        "            reference_group = next((prepare_data_frame(group) for (source, signal), group in by_signal if source == reference.source and signal == reference.signal[0]), None)\n",
        "\n",
        "            if reference_group is None or reference_group.empty:\n",
        "                continue  # no data for reference\n",
        "\n",
        "            # dataframe structure: index=time_value, value=value\n",
        "            other_groups = [((source, signal), prepare_data_frame(group)) for (source, signal), group in by_signal if not (source == reference.source and signal == reference.signal[0])]\n",
        "            if not other_groups:\n",
        "                continue  # no other signals\n",
        "\n",
        "            for (source, signal), other_group in other_groups:\n",
        "                for cor in compute_correlations(geo_type, geo_value, source, signal, lag, reference_group, other_group):\n",
        "                    yield cor.asdict()\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(filter_fields(gen()))\n",
        "\n",
        "\n",
        "@bp.route(\"/csv\", methods=(\"GET\", \"POST\"))\n",
        "def handle_export():\n",
        "    source, signal = request.args.get(\"signal\", \"jhu-csse:confirmed_incidence_num\").split(\":\")\n",
        "    start_day = request.args.get(\"start_day\", \"2020-04-01\")\n",
        "    end_day = request.args.get(\"end_day\", \"2020-09-01\")\n",
        "    geo_type = request.args.get(\"geo_type\", \"county\")\n",
        "    geo_values = request.args.get(\"geo_values\", \"*\")\n",
        "\n",
        "    if geo_values != \"*\":\n",
        "        geo_values = geo_values.split(\",\")\n",
        "\n",
        "    as_of = request.args.get(\"as_of\", None)\n",
        "\n",
        "    start_day = datetime.strptime(start_day, \"%Y-%m-%d\").date()\n",
        "    end_day = datetime.strptime(end_day, \"%Y-%m-%d\").date()\n",
        "\n",
        "    if as_of is not None:\n",
        "        as_of = datetime.strptime(as_of, \"%Y-%m-%d\").date()\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"datapoint\", \"t\")\n",
        "\n",
        "    q.set_fields([\"geo_value\", \"signal\", \"time_value\", \"issue\", \"lag\", \"value\", \"stderr\", \"sample_size\", \"geo_type\", \"source\"], [], [])\n",
        "    q.set_order(\"time_value\", \"geo_value\")\n",
        "    q.where(source=source, signal=signal, time_type=\"day\")\n",
        "    q.conditions.append(\"time_value BETWEEN :start_day AND :end_day\")\n",
        "    q.params[\"start_day\"] = date_to_time_value(start_day)\n",
        "    q.params[\"end_day\"] = date_to_time_value(end_day)\n",
        "    q.where_geo_pairs(\"geo_type\", \"geo_value\", [GeoPair(geo_type, True if geo_values == \"*\" else geo_values)])\n",
        "\n",
        "    _handle_lag_issues_as_of(q, None, None, date_to_time_value(as_of) if as_of is not None else None)\n",
        "\n",
        "    # tag as_of in filename, if it was specified\n",
        "    as_of_str = \"-asof-{as_of}\".format(as_of=as_of.isoformat()) if as_of is not None else \"\"\n",
        "    filename = \"covidcast-{source}-{signal}-{start_day}-to-{end_day}{as_of}\".format(source=source, signal=signal, start_day=start_day.isoformat(), end_day=end_day.isoformat(), as_of=as_of_str)\n",
        "    p = CSVPrinter(filename)\n",
        "\n",
        "    def parse_row(i, row):\n",
        "        # '',geo_value,signal,{time_value,issue},lag,value,stderr,sample_size,geo_type,data_source\n",
        "        return {\n",
        "            \"\": i,\n",
        "            \"geo_value\": row[\"geo_value\"],\n",
        "            \"signal\": row[\"signal\"],\n",
        "            \"time_value\": time_value_to_iso(row[\"time_value\"]),\n",
        "            \"issue\": time_value_to_iso(row[\"issue\"]),\n",
        "            \"lag\": row[\"lag\"],\n",
        "            \"value\": row[\"value\"],\n",
        "            \"stderr\": row[\"stderr\"],\n",
        "            \"sample_size\": row[\"sample_size\"],\n",
        "            \"geo_type\": row[\"geo_type\"],\n",
        "            \"data_source\": row[\"source\"],\n",
        "        }\n",
        "\n",
        "    def gen(first_row, rows):\n",
        "        yield parse_row(0, first_row)\n",
        "        for i, row in enumerate(rows):\n",
        "            yield parse_row(i + 1, row)\n",
        "\n",
        "    # execute query\n",
        "    try:\n",
        "        r = run_query(p, (str(q), q.params))\n",
        "    except Exception as e:\n",
        "        raise DatabaseErrorException(str(e))\n",
        "\n",
        "    # special case for no data to be compatible with the CSV server\n",
        "    first_row = next(r, None)\n",
        "    if not first_row:\n",
        "        return \"No matching data found for signal {source}:{signal} \" \"at {geo} level from {start_day} to {end_day}, as of {as_of}.\".format(\n",
        "            source=source, signal=signal, geo=geo_type, start_day=start_day.isoformat(), end_day=end_day.isoformat(), as_of=(date.today().isoformat() if as_of is None else as_of.isoformat())\n",
        "        )\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(gen(first_row, r))\n",
        "\n",
        "\n",
        "@bp.route(\"/backfill\", methods=(\"GET\", \"POST\"))\n",
        "def handle_backfill():\n",
        "    \"\"\"\n",
        "    example query: http://localhost:5000/covidcast/backfill?signal=fb-survey:smoothed_cli&time=day:20200101-20220101&geo=state:ny&anchor_lag=60\n",
        "    \"\"\"\n",
        "    require_all(\"geo\", \"time\", \"signal\")\n",
        "    signal_pair = parse_single_source_signal_arg(\"signal\")\n",
        "    time_pair = parse_single_time_arg(\"time\")\n",
        "    geo_pair = parse_single_geo_arg(\"geo\")\n",
        "    reference_anchor_lag = extract_integer(\"anchor_lag\")  # in days\n",
        "    if reference_anchor_lag is None:\n",
        "        reference_anchor_lag = 60\n",
        "\n",
        "    # build query\n",
        "    q = QueryBuilder(\"datapoint\", \"t\")\n",
        "\n",
        "    fields_string = []\n",
        "    fields_int = [\"time_value\", \"issue\"]\n",
        "    fields_float = [\"value\", \"sample_size\"]\n",
        "    # sort by time value and issue asc\n",
        "    q.set_order(time_value=True, issue=True)\n",
        "    q.set_fields(fields_string, fields_int, fields_float, [\"is_latest_issue\"])\n",
        "\n",
        "    q.where_source_signal_pairs(\"source\", \"signal\", [signal_pair])\n",
        "    q.where_geo_pairs(\"geo_type\", \"geo_value\", [geo_pair])\n",
        "    q.where_time_pairs(\"time_type\", \"time_value\", [time_pair])\n",
        "\n",
        "    # no restriction of issues or dates since we want all issues\n",
        "    # _handle_lag_issues_as_of(q, issues, lag, as_of)\n",
        "\n",
        "    p = create_printer()\n",
        "\n",
        "    def find_anchor_row(rows: List[Dict[str, Any]], issue: int) -> Optional[Dict[str, Any]]:\n",
        "        # assume sorted by issue asc\n",
        "        # find the row that is <= target issue\n",
        "        i = bisect_right([r[\"issue\"] for r in rows], issue)\n",
        "        if i:\n",
        "            return rows[i - 1]\n",
        "        return None\n",
        "\n",
        "    def gen(rows):\n",
        "        # stream per time_value\n",
        "        for time_value, group in groupby((parse_row(row, fields_string, fields_int, fields_float) for row in rows), lambda row: row[\"time_value\"]):\n",
        "            # compute data per time value\n",
        "            issues: List[Dict[str, Any]] = [r for r in group]\n",
        "            anchor_row = find_anchor_row(issues, shift_time_value(time_value, reference_anchor_lag))\n",
        "\n",
        "            for i, row in enumerate(issues):\n",
        "                if i > 0:\n",
        "                    prev_row = issues[i - 1]\n",
        "                    row[\"value_rel_change\"] = compute_trend_value(row[\"value\"] or 0, prev_row[\"value\"] or 0, 0)\n",
        "                    if row[\"sample_size\"] is not None:\n",
        "                        row[\"sample_size_rel_change\"] = compute_trend_value(row[\"sample_size\"] or 0, prev_row[\"sample_size\"] or 0, 0)\n",
        "                if anchor_row and anchor_row[\"value\"] is not None:\n",
        "                    row[\"is_anchor\"] = row == anchor_row\n",
        "                    row[\"value_completeness\"] = (row[\"value\"] or 0) / anchor_row[\"value\"] if anchor_row[\"value\"] else 1\n",
        "                    if row[\"sample_size\"] is not None:\n",
        "                        row[\"sample_size_completeness\"] = row[\"sample_size\"] / anchor_row[\"sample_size\"] if anchor_row[\"sample_size\"] else 1\n",
        "                yield row\n",
        "\n",
        "    # execute first query\n",
        "    try:\n",
        "        r = run_query(p, (q.query, q.params))\n",
        "    except Exception as e:\n",
        "        raise DatabaseErrorException(str(e))\n",
        "\n",
        "    # now use a generator for sending the rows and execute all the other queries\n",
        "    return p(filter_fields(gen(r)))\n",
        "\n",
        "\n",
        "@bp.route(\"/meta\", methods=(\"GET\", \"POST\"))\n",
        "def handle_meta():\n",
        "    \"\"\"\n",
        "    similar to /covidcast_meta but in a structured optimized JSON form for the app\n",
        "    \"\"\"\n",
        "\n",
        "    signal = parse_source_signal_arg(\"signal\")\n",
        "\n",
        "    row = db.execute(text(\"SELECT epidata FROM covidcast_meta_cache LIMIT 1\")).fetchone()\n",
        "\n",
        "    data = loads(row[\"epidata\"]) if row and row[\"epidata\"] else []\n",
        "\n",
        "    all_signals: AllSignalsMap = {}\n",
        "    for row in data:\n",
        "        if row[\"time_type\"] != \"day\":\n",
        "            continue\n",
        "        entry: Set[str] = all_signals.setdefault(row[\"data_source\"], set())\n",
        "        entry.add(row[\"signal\"])\n",
        "\n",
        "    out: Dict[str, CovidcastMetaEntry] = {}\n",
        "    for row in data:\n",
        "        if row[\"time_type\"] != \"day\":\n",
        "            continue\n",
        "        if signal and all((not s.matches(row[\"data_source\"], row[\"signal\"]) for s in signal)):\n",
        "            continue\n",
        "        entry = out.setdefault(\n",
        "            f\"{row['data_source']}:{row['signal']}\", CovidcastMetaEntry(row[\"data_source\"], row[\"signal\"], row[\"min_time\"], row[\"max_time\"], row[\"max_issue\"], {}, all_signals=all_signals)\n",
        "        )\n",
        "        entry.intergrate(row)\n",
        "\n",
        "    return jsonify([r.asdict() for r in out.values()])\n",
        "\n",
        "\n",
        "@bp.route(\"/coverage\", methods=(\"GET\", \"POST\"))\n",
        "def handle_coverage():\n",
        "    \"\"\"\n",
        "    similar to /signal_dashboard_coverage for a specific signal returns the coverage (number of locations for a given geo_type)\n",
        "    \"\"\"\n",
        "\n",
        "    signal = parse_source_signal_pairs()\n",
        "    geo_type = request.args.get(\"geo_type\", \"county\")\n",
        "    if \"window\" in request.values:\n",
        "        time_window = parse_day_range_arg(\"window\")\n",
        "    else:\n",
        "        now_time = extract_date(\"latest\")\n",
        "        now = date.today() if now_time is None else time_value_to_date(now_time)\n",
        "        last = extract_integer(\"days\")\n",
        "        if last is None:\n",
        "            last = 30\n",
        "        time_window = (date_to_time_value(now - timedelta(days=last)), date_to_time_value(now))\n",
        "\n",
        "    q = QueryBuilder(\"covidcast\", \"c\")\n",
        "    fields_string = [\"source\", \"signal\"]\n",
        "    fields_int = [\"time_value\"]\n",
        "\n",
        "    q.set_fields(fields_string, fields_int)\n",
        "\n",
        "    # manually append the count column because of grouping\n",
        "    fields_int.append(\"count\")\n",
        "    q.fields.append(f\"count({q.alias}.geo_value) as count\")\n",
        "\n",
        "    if geo_type == \"only-county\":\n",
        "        q.where(geo_type=\"county\")\n",
        "        q.conditions.append('geo_value not like \"%000\"')\n",
        "    else:\n",
        "        q.where(geo_type=geo_type)\n",
        "    q.where_source_signal_pairs(\"source\", \"signal\", signal)\n",
        "    q.where_time_pairs(\"time_type\", \"time_value\", [TimePair(\"day\", [time_window])])\n",
        "    q.group_by = \"c.source, c.signal, c.time_value\"\n",
        "    q.set_order(\"source\", \"signal\", \"time_value\")\n",
        "\n",
        "    _handle_lag_issues_as_of(q, None, None, None)\n",
        "\n",
        "    return execute_query(q.query, q.params, fields_string, fields_int, [])\n",
        "\n",
        "\n",
        "@bp.route(\"/anomalies\", methods=(\"GET\", \"POST\"))\n",
        "def handle_anomalies():\n",
        "    \"\"\"\n",
        "    proxy to the excel sheet about data anomalies\n",
        "    \"\"\"\n",
        "\n",
        "    signal = parse_source_signal_arg(\"signal\")\n",
        "\n",
        "    df = read_csv(\n",
        "        \"https://docs.google.com/spreadsheets/d/e/2PACX-1vToGcf9x5PNJg-eSrxadoR5b-LM2Cqs9UML97587OGrIX0LiQDcU1HL-L2AA8o5avbU7yod106ih0_n/pub?gid=0&single=true&output=csv\", skip_blank_lines=True\n",
        "    )\n",
        "    df = df[df[\"source\"].notnull() & df[\"published\"]]\n",
        "    return print_pandas(df)\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "error",
          "ename": "ImportError",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
            "\u001b[0;31mImportError\u001b[0m                               Traceback (most recent call last)",
            "\u001b[0;32m<ipython-input-2-e2109d7cce20>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      8\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpandas\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mread_csv\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      9\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 10\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_common\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mis_compatibility_mode\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdb\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     11\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exceptions\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mValidationFailedException\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mDatabaseErrorException\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     12\u001b[0m from .._params import (\n",
            "\u001b[0;31mImportError\u001b[0m: attempted relative import with no known parent package",
            "",
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0;32m\nNOTE: If your import is failing due to a missing package, you can\nmanually install dependencies using either !pip or !apt.\n\nTo view examples of installing some common dependencies, click the\n\"Open Examples\" button below.\n\u001b[0;31m---------------------------------------------------------------------------\u001b[0m\n"
          ],
          "errorDetails": {
            "actions": [
              {
                "action": "open_url",
                "actionText": "Open Examples",
                "url": "/notebooks/snippets/importing_libraries.ipynb"
              }
            ]
          }
        }
      ]
    }
  ]
}