code cleanup and lint

Author: Sean Sullivan

elastic_datashader/cache.py

@@ -168,10 +168,10 @@ def age_off_cache(cache_path: Path, idx_name: str, max_age: timedelta) -> None:
     remove_empty_dirs(cache_path/idx_name)
 
 def remove_empty_dirs(path: Path):
-    for root,dirs,_ in os.walk(path, topdown=False):
+    for root, dirs, _ in os.walk(path, topdown=False):
         for d in dirs:
             with suppress(OSError):
-                os.rmdir(Path(root,d))
+                os.rmdir(Path(root, d))
 
 def get_idx_names(cache_path: Path) -> Iterable[str]:
     for path in cache_path.glob("*"):

elastic_datashader/config.py

@@ -18,7 +18,6 @@ class Config:
     cache_cleanup_interval: timedelta
     cache_path: Path
     cache_timeout: timedelta
-    csrf_secret_key: str
     datashader_headers: Dict[Any, Any]
     elastic_hosts: str
     ellipse_render_mode: str
@@ -29,8 +28,6 @@ class Config:
     max_ellipses_per_tile: int
     max_legend_items_per_tile: int
     num_ellipse_points: int
-    proxy_host: Optional[str]
-    proxy_prefix: str
     query_timeout_seconds: int
     render_timeout: timedelta
     tms_key: Optional[str]
@@ -93,7 +90,6 @@ def config_from_env(env) -> Config:
         cache_cleanup_interval=timedelta(seconds=int(env.get("DATASHADER_CACHE_CLEANUP_INTERVAL", 5*60))),
         cache_path=Path(env.get("DATASHADER_CACHE_DIRECTORY", "tms-cache")),
         cache_timeout=timedelta(seconds=int(env.get("DATASHADER_CACHE_TIMEOUT", 60*60))),
-        csrf_secret_key=env.get("DATASHADER_CSRF_SECRET_KEY", "CSRFProtectionKey"),
         datashader_headers=load_datashader_headers(env.get("DATASHADER_HEADER_FILE", "headers.yaml")),
         elastic_hosts=env.get("DATASHADER_ELASTIC", "http://localhost:9200"),
         ellipse_render_mode=env.get("DATASHADER_ELLIPSE_RENDER_MODE", "matrix"),
@@ -104,8 +100,6 @@ def config_from_env(env) -> Config:
         max_ellipses_per_tile=int(env.get("DATASHADER_MAX_ELLIPSES_PER_TILE", 100_000)),
         max_legend_items_per_tile=int(env.get("MAX_LEGEND_ITEMS_PER_TILE", 20)),
         num_ellipse_points=int(env.get("DATASHADER_NUM_ELLIPSE_POINTS", 100)),
-        proxy_host=env.get("DATASHADER_PROXY_HOST", None),
-        proxy_prefix=env.get("DATASHADER_PROXY_PREFIX", ""),
         query_timeout_seconds=int(env.get("DATASHADER_QUERY_TIMEOUT", 0)),
         render_timeout=timedelta(seconds=int(env.get("DATASHADER_RENDER_TIMEOUT", 30))),
         tms_key=env.get("DATASHADER_TMS_KEY", None),

elastic_datashader/elastic.py

@@ -158,26 +158,6 @@ def convert_nm_to_ellipse_units(distance: float, units: str) -> float:
     # NB. assume "majmin_m" if any others
     return distance * 1852
 
-def get_field_type(elastic_hosts: str, headers: Optional[str], params: Dict[str, Any], field: str, idx: str) -> str:
-    user = params.get("user")
-    x_opaque_id = params.get("x-opaque-id")
-    es = Elasticsearch(
-        elastic_hosts.split(","),
-        verify_certs=False,
-        timeout=900,
-        headers=get_es_headers(headers, user, x_opaque_id),
-    )
-    if idx.find("*:") != -1:
-        idx = idx[idx.find("*:")+2:] # when you query for mappings if it is cross cluster you don't get a mapping
-    mappings = es.indices.get_field_mapping(fields=field, index=idx)
-    # {'foot_prints': {'mappings': {'foot_print': {'full_name': 'foot_print', 'mapping': {'foot_print': {'type': 'geo_shape'}}}}}}
-    index = list(mappings.keys())[0] # if index is my_index* it comes back as my_index
-    field_parts = field.split(".")
-    try:
-        return mappings[index]['mappings'][field]['mapping'][field_parts[-1]]['type'] # handles 'geo_center' or a nested object {signal:{geo:{location:{}}}}
-    except AttributeError:
-        return mappings[index]['mappings'][field]['mapping'][field]['type'] # handles literal string with periods 'signal.geo.location'
-
 def get_search_base(
     elastic_hosts: str,
     headers: Optional[str],
@@ -271,21 +251,6 @@ def get_search_base(
 
     return base_s
 
-def handle_range_or_exists_filters(filter_input: Dict[Any, Any]) -> Dict[str, Any]:
-    """
-    `range` and `exists` filters can appear either directly under
-    `filter[]` or under `filter[].query` depending on the version
-    of Kibana, the former being the old way, so they need special
-    handling for backward compatibility.
-    """
-    filter_type = filter_input.get("meta").get("type")  # "range" or "exists"
-
-    # Handle old query structure for backward compatibility
-    if filter_input.get(filter_type) is not None:
-        return {filter_type: filter_input.get(filter_type)}
-
-    return filter_input.get("query")
-
 def build_dsl_filter(filter_inputs) -> Optional[Dict[str, Any]]:
     """
 
@@ -309,7 +274,7 @@ def build_dsl_filter(filter_inputs) -> Optional[Dict[str, Any]]:
             f.get("geo_shape") or
             f.get("geo_distance")
         )
-        if f.get("query",None):
+        if f.get("query", None):
             if f.get("meta").get("negate"):
                 filter_dict["must_not"].append(f.get("query"))
             else:
@@ -318,16 +283,16 @@ def build_dsl_filter(filter_inputs) -> Optional[Dict[str, Any]]:
             if not is_spatial_filter:
                 filt_type = f.get("meta").get("type")
                 if f.get("meta").get("negate"):
-                    filter_dict["must_not"].append({filt_type:f.get(filt_type)})
+                    filter_dict["must_not"].append({filt_type: f.get(filt_type)})
                 else:
-                    filter_dict["filter"].append({filt_type:f.get(filt_type)})
+                    filter_dict["filter"].append({filt_type: f.get(filt_type)})
             else:
-                for geo_type in ["geo_polygon","geo_bounding_box","geo_shape","geo_distance"]:
-                    if f.get(geo_type,None):
+                for geo_type in ["geo_polygon", "geo_bounding_box", "geo_shape", "geo_distance"]:
+                    if f.get(geo_type, None):
                         if f.get("meta").get("negate"):
-                            filter_dict["must_not"].append({geo_type:f.get(geo_type)})
+                            filter_dict["must_not"].append({geo_type: f.get(geo_type)})
                         else:
-                            filter_dict["filter"].append({geo_type:f.get(geo_type)})
+                            filter_dict["filter"].append({geo_type: f.get(geo_type)})
     logger.info("Filter output %s", filter_dict)
     return filter_dict
 
@@ -396,32 +361,6 @@ def parse_duration_interval(interval):
             kwargs[key] = int(interval[0:len(interval)-1])
     return relativedelta(**kwargs)
 
-def convert(response, category_formatter=str):
-    """
-
-    :param response:
-    :return:
-    """
-    if hasattr(response.aggregations, "categories"):
-        for category in response.aggregations.categories:
-            for bucket in category.grids:
-                x, y = lnglat_to_meters(
-                    bucket.centroid.location.lon, bucket.centroid.location.lat
-                )
-                yield {
-                    "lon": bucket.centroid.location.lon,
-                    "lat": bucket.centroid.location.lat,
-                    "x": x,
-                    "y": y,
-                    "c": bucket.centroid.count,
-                    "t": category_formatter(category.key),
-                }
-    else:
-        for bucket in response.aggregations.grids:
-            lon = bucket.centroid.location.lon
-            lat = bucket.centroid.location.lat
-            x, y = lnglat_to_meters(lon, lat)
-            yield {"lon": lon, "lat": lat, "x": x, "y": y, "c": bucket.centroid.count}
 
 def convert_composite(response, categorical, filter_buckets, histogram_interval, category_type, category_format):
     if categorical and filter_buckets is False:
@@ -533,20 +472,6 @@ def get_nested_field_from_hit(hit, field_parts: List[str], default=None):
 
     raise ValueError("field must be provided")
 
-def chunk_iter(iterable, chunk_size):
-    chunks = [None] * chunk_size
-    i = -1
-    for i, v in enumerate(iterable):
-        idx = i % chunk_size
-        if idx == 0 and i > 0:
-            i = -1
-            yield (True, chunks)
-        chunks[idx] = v
-
-    if i >= 0:
-        last_written_idx = i % chunk_size
-        yield (False, chunks[0:last_written_idx+1])
-
 def bucket_noop(bucket, search):
     # pylint: disable=unused-argument
     return bucket

elastic_datashader/parameters.py

@@ -1,7 +1,6 @@
 from datetime import datetime, timedelta, timezone
 from hashlib import sha256
 from json import loads
-from socket import gethostname
 from time import sleep
 from typing import Any, Dict, Optional, Tuple
 from urllib.parse import unquote
@@ -351,7 +350,6 @@ def generate_global_params(headers, params, idx):
     if category_type == "number":
         bounds_s.aggs.metric("field_stats", "stats", field=category_field)
 
-    # field_type = get_field_type(config.elastic_hosts, headers, params, geopoint_field, idx)
     field_type = params["geofield_type"] # CCS you cannot get mappings so we needed to push the field type from the client side
     # Execute and process search
     if len(list(bounds_s.aggs)) > 0 and field_type != "geo_shape":
@@ -470,7 +468,7 @@ def generate_global_params(headers, params, idx):
 
 
 def merge_generated_parameters(headers, params, idx, param_hash):
-    layer_id = f"{param_hash}_{gethostname()}"
+    layer_id = f"{param_hash}_{config.hostname}"
     es = Elasticsearch(
         config.elastic_hosts.split(","),
         verify_certs=False,
@@ -488,7 +486,7 @@ def merge_generated_parameters(headers, params, idx, param_hash):
         try:
             doc = Document(
                 _id=layer_id,
-                creating_host=gethostname(),
+                creating_host=config.hostname,
                 creating_pid=os.getpid(),
                 creating_timestamp=datetime.now(timezone.utc),
                 generated_params=None,
@@ -533,7 +531,7 @@ def merge_generated_parameters(headers, params, idx, param_hash):
             generated_params = {
                 "complete": False,
                 "generation_start_time": datetime.now(timezone.utc),
-                "generating_host": gethostname(),
+                "generating_host": config.hostname,
                 "generating_pid": os.getpid(),
             }
 

elastic_datashader/routers/tms.py

@@ -25,7 +25,7 @@
 )
 from ..config import config
 from ..drawing import generate_x_tile
-from ..elastic import get_es_headers,get_search_base
+from ..elastic import get_es_headers, get_search_base
 from ..logger import logger
 from ..parameters import extract_parameters, merge_generated_parameters
 from ..tilegen import (
@@ -297,7 +297,7 @@ async def fetch_or_render_tile(already_waited: int, idx: str, x: int, y: int, z:
         # try to build the dsl object bad filters cause exceptions that are then retried.
         # underlying elasticsearch_dsl doesn't support the elasticsearch 8 api yet so this causes requests to thrash
         # If the filters are bad or elasticsearch_dsl cannot build the request will never be completed so serve X tile
-        get_search_base(config.elastic_hosts,request.headers,params,idx)
+        get_search_base(config.elastic_hosts, request.headers, params, idx)
     except Exception as ex:  # pylint: disable=W0703
         logger.exception("Error while extracting parameters")
         params = {"user": request.headers.get("es-security-runas-user", None)}

elastic_datashader/tilegen.py

@@ -1099,7 +1099,6 @@ def generate_tile(idx, x, y, z, headers, params, tile_width_px=256, tile_height_
 
         # the composite needs one bin for 'after_key'
         composite_agg_size = int(max_bins / inner_agg_size) - 1
-        # field_type = get_field_type(config.elastic_hosts, headers, params, geopoint_field, idx)
         field_type = params["geofield_type"] # CCS you cannot get mappings so we needed to push the field type from the client side
         partial_data = False # TODO can we get partial data?
         span = None

tests/test_config.py

@@ -15,8 +15,6 @@ def test_config_defaults():
     assert cfg.cache_path == Path("tms-cache")
     assert cfg.cache_timeout == timedelta(seconds=3600)
     assert cfg.elastic_hosts == "http://localhost:9200"
-    assert cfg.proxy_host is None
-    assert cfg.proxy_prefix == ""
     assert cfg.tms_key is None
     assert cfg.max_bins == 10000
     assert cfg.max_batch == 10000
@@ -56,8 +54,6 @@ def test_config_env():
     assert cfg.cache_path == Path("tms-cache-foo")
     assert cfg.cache_timeout == timedelta(seconds=60)
     assert cfg.elastic_hosts == "http://localhost:9201"
-    assert cfg.proxy_host == "http://localhost:1337"
-    assert cfg.proxy_prefix == "foo"
     assert cfg.tms_key == "bar"
     assert cfg.max_bins == 10
     assert cfg.max_batch == 1000

tests/test_elastic.py

@@ -152,66 +152,3 @@ def test_split_fieldname_to_list(field, expected):
 def test_get_nested_field_from_hit():
     pass
 
-@pytest.mark.parametrize(
-    "filter_input,filter_type,new_way,expected",
-    (
-        ({"meta": {"type": "exists"}, "query": {"exists": {"field": "foo"}}}, "exists", True, {"exists": {"field": "foo"}}),
-        ({"meta": {"type": "range"}, "range": {"from": "foo", "to": "bar"}}, "range", False, {"range": {"from": "foo", "to": "bar"}}),
-    )
-)
-def test_handle_range_or_exists_filters(filter_input, filter_type, new_way, expected):
-    filter_output = elastic.handle_range_or_exists_filters(filter_input)
-    assert len(filter_output) == 1
-    assert filter_type in filter_output
-    assert type(filter_output[filter_type]) is dict
-
-    if new_way:
-        expected_output = filter_input["query"]
-    else:
-        expected_output = {filter_type: filter_input[filter_type]}
-
-    for key in expected_output:
-        assert key in filter_output
-
-        for subkey in expected_output[key]:
-            assert subkey in filter_output[key]
-            assert expected_output[key][subkey] == filter_output[key][subkey]
-
-def test_chunk_iter():
-    for has_more, chunk in elastic.chunk_iter([], 1000):
-        assert True == False
-
-    for has_more, chunk in elastic.chunk_iter(range(10), 1000):
-        assert has_more ==  False
-        assert len(chunk) == 10
-
-    for has_more, chunk in elastic.chunk_iter(range(1000), 1000):
-        assert has_more ==  False
-        assert len(chunk) == 1000
-
-    for ii, (has_more, chunk) in enumerate(elastic.chunk_iter(range(1001), 1000)):
-        if ii == 0:
-            assert has_more ==  True
-            assert len(chunk) == 1000
-        elif ii == 1:
-            assert has_more ==  False
-            assert len(chunk) == 1
-
-    for ii, (has_more, chunk) in enumerate(elastic.chunk_iter(range(2000), 1000)):
-        if ii == 0:
-            assert has_more ==  True
-            assert len(chunk) == 1000
-        elif ii == 1:
-            assert has_more ==  False
-            assert len(chunk) == 1000
-
-    for ii, (has_more, chunk) in enumerate(elastic.chunk_iter(range(2010), 1000)):
-        if ii == 0:
-            assert has_more ==  True
-            assert len(chunk) == 1000
-        elif ii == 1:
-            assert has_more ==  True
-            assert len(chunk) == 1000
-        elif ii == 2:
-            assert has_more ==  False
-            assert len(chunk) == 10