Merge pull request #292 from investigativedata/develop

v0.6.10

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.6.9
+current_version = 0.6.10
 commit = True
 tag = True
 message = 🔖 Bump version: {current_version} → {new_version}

VERSION

@@ -1 +1 @@
-0.6.9
+0.6.10

ftmq/__init__.py

@@ -1,4 +1,4 @@
 from ftmq.query import Query
 
-__version__ = "0.6.9"
+__version__ = "0.6.10"
 __all__ = ["Query"]

ftmq/cli.py

@@ -8,13 +8,14 @@
 
 from ftmq.aggregate import aggregate
 from ftmq.io import apply_datasets, smart_read_proxies, smart_write_proxies
+from ftmq.logging import get_logger
 from ftmq.model.coverage import Collector
 from ftmq.model.dataset import Catalog, Dataset
 from ftmq.query import Query
 from ftmq.store import get_store
 from ftmq.util import parse_unknown_filters
 
-log = logging.getLogger(__name__)
+log = get_logger(__name__)
 
 
 @click.group(cls=DefaultGroup, default="q", default_if_no_args=True)

ftmq/dedupe.py

@@ -1,11 +1,12 @@
-import logging
 from functools import cache
 
 from anystore import smart_stream
 from nomenklatura.entity import CompositeEntity
 from nomenklatura.resolver import Edge, Resolver
 
-log = logging.getLogger(__name__)
+from ftmq.logging import get_logger
+
+log = get_logger(__name__)
 
 
 @cache

ftmq/io.py

@@ -1,5 +1,4 @@
-import logging
-from collections.abc import Iterable
+from typing import Iterable
 
 import orjson
 from anystore.io import smart_open, smart_stream
@@ -8,12 +7,13 @@
 from nomenklatura.util import PathLike
 
 from ftmq.exceptions import ValidationError
+from ftmq.logging import get_logger
 from ftmq.query import Query
 from ftmq.store import Store, get_store
 from ftmq.types import CEGenerator, SDict
 from ftmq.util import get_statements, make_dataset, make_proxy
 
-log = logging.getLogger(__name__)
+log = get_logger(__name__)
 
 DEFAULT_MODE = "rb"
 

ftmq/logging.py

@@ -0,0 +1,103 @@
+import logging
+import sys
+from logging import Filter, LogRecord
+from typing import Any, Dict, List
+
+import structlog
+from structlog.contextvars import merge_contextvars
+from structlog.dev import ConsoleRenderer, set_exc_info
+from structlog.processors import (
+    JSONRenderer,
+    TimeStamper,
+    UnicodeDecoder,
+    add_log_level,
+    format_exc_info,
+)
+from structlog.stdlib import (
+    BoundLogger,
+    LoggerFactory,
+    ProcessorFormatter,
+    add_logger_name,
+)
+from structlog.stdlib import get_logger as get_raw_logger
+
+LOG_JSON = False
+LOG_LEVEL = logging.INFO
+
+
+def get_logger(name: str, *args, **kwargs) -> BoundLogger:
+    return get_raw_logger(name, *args, **kwargs)
+
+
+def configure_logging(level: int = logging.INFO) -> None:
+    """Configure log levels and structured logging"""
+    shared_processors: List[Any] = [
+        add_log_level,
+        add_logger_name,
+        # structlog.stdlib.PositionalArgumentsFormatter(),
+        # structlog.processors.StackInfoRenderer(),
+        merge_contextvars,
+        set_exc_info,
+        TimeStamper(fmt="iso"),
+        # format_exc_info,
+        UnicodeDecoder(),
+    ]
+
+    if LOG_JSON:
+        shared_processors.append(format_exc_info)
+        shared_processors.append(format_json)
+        formatter = ProcessorFormatter(
+            foreign_pre_chain=shared_processors,
+            processor=JSONRenderer(),
+        )
+    else:
+        formatter = ProcessorFormatter(
+            foreign_pre_chain=shared_processors,
+            processor=ConsoleRenderer(
+                exception_formatter=structlog.dev.plain_traceback
+            ),
+        )
+
+    processors = shared_processors + [
+        ProcessorFormatter.wrap_for_formatter,
+    ]
+
+    # configuration for structlog based loggers
+    structlog.configure(
+        cache_logger_on_first_use=True,
+        # wrapper_class=AsyncBoundLogger,
+        wrapper_class=BoundLogger,
+        processors=processors,
+        context_class=dict,
+        logger_factory=LoggerFactory(),
+    )
+
+    # handler for low level logs that should be sent to STDOUT
+    out_handler = logging.StreamHandler(sys.stdout)
+    out_handler.setLevel(level)
+    out_handler.addFilter(_MaxLevelFilter(logging.WARNING))
+    out_handler.setFormatter(formatter)
+    # handler for high level logs that should be sent to STDERR
+    error_handler = logging.StreamHandler(sys.stderr)
+    error_handler.setLevel(logging.ERROR)
+    error_handler.setFormatter(formatter)
+
+    root_logger = logging.getLogger()
+    root_logger.setLevel(LOG_LEVEL)
+    root_logger.addHandler(out_handler)
+    root_logger.addHandler(error_handler)
+
+
+def format_json(_: Any, __: Any, ed: Dict[str, str]) -> Dict[str, str]:
+    """Stackdriver uses `message` and `severity` keys to display logs"""
+    ed["message"] = ed.pop("event")
+    ed["severity"] = ed.pop("level", "info").upper()
+    return ed
+
+
+class _MaxLevelFilter(Filter):
+    def __init__(self, highest_log_level: int) -> None:
+        self._highest_log_level = highest_log_level
+
+    def filter(self, log_record: LogRecord) -> bool:
+        return log_record.levelno <= self._highest_log_level

ftmq/query.py

@@ -177,6 +177,21 @@ def dataset_names(self) -> set[str]:
     def schemata(self) -> set[SchemaFilter]:
         return {f for f in self.filters if isinstance(f, SchemaFilter)}
 
+    @property
+    def schemata_names(self) -> set[str]:
+        names = set()
+        for f in self.schemata:
+            names.update(ensure_list(f.value))
+        return names
+
+    @property
+    def countries(self) -> set[str]:
+        names = set()
+        for f in self.properties:
+            if f.key == "country":
+                names.update(ensure_list(f.value))
+        return names
+
     @property
     def reversed(self) -> set[ReverseFilter]:
         return {f for f in self.filters if isinstance(f, ReverseFilter)}

ftmq/store/base.py

@@ -1,17 +1,17 @@
-import logging
 from typing import Iterable
 
 from nomenklatura import store as nk
 from nomenklatura.resolver import Resolver
 
 from ftmq.aggregations import AggregatorResult
+from ftmq.logging import get_logger
 from ftmq.model.coverage import Collector, DatasetStats
 from ftmq.model.dataset import C, Dataset
 from ftmq.query import Q
 from ftmq.types import CE, CEGenerator
 from ftmq.util import DefaultDataset, ensure_dataset, make_dataset
 
-log = logging.getLogger(__name__)
+log = get_logger(__name__)
 
 
 class Store(nk.Store):
@@ -76,10 +76,12 @@ def entities(self, query: Q | None = None) -> CEGenerator:
         else:
             yield from view.entities()
 
-    def get_adjacents(self, proxies: Iterable[CE]) -> set[CE]:
+    def get_adjacents(
+        self, proxies: Iterable[CE], inverted: bool | None = False
+    ) -> set[CE]:
         seen: set[CE] = set()
         for proxy in proxies:
-            for _, adjacent in self.get_adjacent(proxy):
+            for _, adjacent in self.get_adjacent(proxy, inverted=inverted):
                 if adjacent.id not in seen:
                     seen.add(adjacent)
         return seen

ftmq/util.py

@@ -3,7 +3,7 @@
 from typing import Any, Generator
 
 import pycountry
-from banal import ensure_list
+from banal import ensure_list, is_listish
 from followthemoney.schema import Schema
 from followthemoney.types import registry
 from followthemoney.util import make_entity_id, sanitize_text
@@ -57,7 +57,11 @@ def parse_unknown_filters(
         prop, *op = prop.split("__")
         op = op[0] if op else Comparators.eq
         if op == Comparators["in"]:
-            value = value.split(",")
+            # de,fr or ["de", "fr"]
+            if is_listish(value):
+                value = ensure_list(value)
+            else:
+                value = value.split(",")
         yield prop, value, op
 
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@investigativedata/ftmq",
-  "version": "0.6.9",
+  "version": "0.6.10",
   "description": "javascript interface for ftmq",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",