wipwip

Author: aaxelb

CONTRIBUTING.md

@@ -105,5 +105,5 @@ pc up testbox
 
 example devloop -- build with current code, lint and run tests with debugger on error, stop on failure
 ```
-pc run --build --rm --no-deps testbox poetry run python -m elasticsearch_metrics.tests --devloop
+pc run --build --rm --no-deps testbox poetry run python -m elasticsearch_metrics.tests --failfast --pdb
 ```

elasticsearch_metrics/imps/elastic8.py

@@ -27,12 +27,16 @@
 from elasticsearch8.exceptions import NotFoundError
 from elasticsearch8 import Elasticsearch as Elastic8Client
 from elasticsearch8 import dsl as esdsl
-from elasticsearch8._sync.document import IndexMeta
+from elasticsearch8.dsl._sync.document import IndexMeta
 
 from elasticsearch_metrics import signals
 from elasticsearch_metrics import exceptions
 from elasticsearch_metrics.registry import djelme_registry
-from elasticsearch_metrics.protocols import ProtoDjelmeBackend, ProtoCountedUsage
+from elasticsearch_metrics.protocols import (
+    ProtoDjelmeBackend,
+    ProtoCountedUsage,
+    ProtoDjelmeRecord,
+)
 from elasticsearch_metrics.util import timeseries_naming
 from elasticsearch_metrics.util.unique_together import get_unique_id
 from elasticsearch_metrics.util.anon_enough import opaque_sessionhour_id
@@ -144,14 +148,16 @@ class Meta:
         abstract = True
 
     @classmethod
-    def record(cls, *, using=None, **kwargs):
+    def record(
+        cls, *, using: str | None = None, **kwargs: typing.Any
+    ) -> "typing.Self":  # typing.Self added in py 3.11 -- str annotation until 3.10 eol
         """Persist a record in Elasticsearch."""
         _instance = cls(**kwargs)
         _instance.save(using=using)
         return _instance
 
     @classmethod
-    def check_djelme_setup(cls, using: str | Elastic8Client | None = None) -> bool:
+    def check_djelme_setup(cls, using: str | None = None) -> bool:
         return bool(cls._index.get(using=using))
 
     @classmethod
@@ -225,7 +231,11 @@ class TimeseriesRecord(DjelmeRecordtype):
     timestamp: datetime.datetime = esdsl.mapped_field(
         default_factory=lambda: django.utils.timezone.now()
     )
-    timeparts: str = esdsl.mapped_field(esdsl.Version())
+    # the 'version' field type allows range queries on semver-like strings
+    # that fit perfectly with "timeparts" representation of a UTC datetime
+    # as a sequence of integers -- use to avoid time zones and date math
+    # (e.g. '2000' < '2000.5.10' < '2000.5.20.20.20' < '2000.11' < '2001')
+    timestamp_parts: str = esdsl.mapped_field(esdsl.Version(), default="")
 
     class Meta:
         abstract = True
@@ -253,11 +263,16 @@ def search_timeseries_range(
         from_when: tuple[int, ...] | datetime.date,
         until_when: tuple[int, ...] | datetime.date,
         **kwargs: typing.Any,
-    ) -> esdsl.Search:
+    ) -> typing.Any:
         _index_pattern = cls.format_timeseries_index_pattern_for_range(
             from_when, until_when
         )
-        _timestamp_q = esdsl.query.Range("timestamp", gte=..., lt=...)
+        _timedepth = cls.get_timedepth()
+        _timestamp_q = esdsl.query.Range(
+            "timestamp_parts",
+            gte=timeseries_naming.semverlike_timeparts(from_when, timedepth=_timedepth),
+            lt=timeseries_naming.semverlike_timeparts(until_when, timedepth=_timedepth),
+        )
         return cls.search(index=_index_pattern).filter(_timestamp_q)
 
     @classmethod
@@ -425,6 +440,13 @@ def check_djelme_setup(cls, using: str | Elastic8Client | None = None) -> bool:
     ###
     # instance methods
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.timestamp_parts = self._build_timestamp_parts()
+
+    def _build_timestamp_parts(self) -> str:
+        return timeseries_naming.full_semverlike_timeparts(self.timestamp)
+
     def djelme_index_name(self) -> str:
         assert self.timestamp is not None
         return self.format_timeseries_index_name(self.timestamp)
@@ -567,6 +589,7 @@ def djelme_teardown(self, recordtypes: collections.abc.Iterable[type]) -> None:
     # for static type-checking; verify intent
     _: type[ProtoCountedUsage] = CountedUsageRecord
     __: type[ProtoDjelmeBackend] = DjelmeElastic8Backend
+    ___: type[ProtoDjelmeRecord] = DjelmeRecordtype
 
 ###
 # names expected by ProtoDjelmeImp

elasticsearch_metrics/protocols.py

@@ -44,15 +44,13 @@ def djelme_teardown(self, recordtypes: collections.abc.Iterable[type]) -> None:
 class ProtoDjelmeRecord(typing.Protocol):
     @classmethod
     def record(
-        cls, **kwargs: typing.Any
+        cls, *, using: str | None = None, **kwargs: typing.Any
     ) -> "typing.Self":  # typing.Self added in py 3.11 -- str annotation until 3.10 eol
         ...
 
     @classmethod
     def check_djelme_setup(cls, using: str | None = None) -> bool: ...
 
-    def djelme_index_name(self) -> str: ...
-
     @classmethod
     def search_timeseries_range(
         cls,
@@ -64,6 +62,8 @@ def search_timeseries_range(
     # @classmethod
     # def each_timeseries_index_status(cls) -> collections.abc.Iterable[str]: ...
 
+    def djelme_index_name(self) -> str: ...
+
 
 ###
 # counter?

elasticsearch_metrics/tests/_test_util.py

@@ -78,5 +78,5 @@ def setUp(self):
             mock.patch("elasticsearch_metrics.imps.elastic6.Document.save"),
         )
         self.mocked_es8_save = self.enterContext(
-            mock.patch("elasticsearch_metrics.imps.elastic8.Document.save"),
+            mock.patch("elasticsearch_metrics.imps.elastic8.esdsl.Document.save"),
         )

elasticsearch_metrics/util/timeseries_naming.py

@@ -13,6 +13,7 @@
     "parse_index_pattern",
     "format_namepart",
     "format_template_name",
+    "timeparts_from_date",
 )
 
 import collections
@@ -103,10 +104,8 @@ def _each_timeparts_for_timerange(
     if timedepth <= 0:
         yield (), True  # reached timedepth; wildcard
         return
-    _from_part, *_from_rest = from_timeparts
-    _until_part, *_until_rest = until_timeparts
-    assert _from_part <= _until_part
-
+    _from_part, *_from_rest = from_timeparts or [0]
+    _until_part, *_until_rest = until_timeparts or [0]
     if include_less_granular:
         yield (), False  # include less-granular non-wildcard index, if it exists
     if _from_part == _until_part:
@@ -118,7 +117,7 @@ def _each_timeparts_for_timerange(
             include_less_granular=include_less_granular,
         ):
             yield (_from_part, *_rest_parts), _is_wildcard
-    elif (_until_part - _from_part) <= max_fanout:  # not too far apart
+    elif 0 < (_until_part - _from_part) <= max_fanout:  # not too far apart
         for _parallel_part in range(_from_part, _until_part):
             yield (_parallel_part,), True  # wildcard
         if any(_until_rest):  # some of the "until" bucket is included
@@ -250,22 +249,13 @@ def format_index_pattern_for_range(
     >>> format_index_pattern_for_range('ap', 'rt',
     ...     (200, 5), datetime.date(200, 5, 8),
     ...     timedepth=3)
+    'ap_rt_200_05_*'
     """
-    _from_timeparts = (
-        timeparts_from_date(from_when, timedepth)
-        if isinstance(from_when, datetime.date)
-        else from_when
-    )
-    _until_timeparts = (
-        timeparts_from_date(until_when, timedepth)
-        if isinstance(until_when, datetime.date)
-        else until_when
-    )
     return format_index_pattern_for_timerange(
-        prefix,
-        recordtype,
-        _from_timeparts,
-        _until_timeparts,
+        prefix=prefix,
+        recordtype=recordtype,
+        from_timeparts=_whenparts(from_when, timedepth),
+        until_timeparts=_whenparts(until_when, timedepth),
         timedepth=timedepth,
         include_less_granular=include_less_granular,
         only_datelike=True,
@@ -365,7 +355,7 @@ def _parse_timename(given_timename: str) -> Iterator[int]:
         yield int(_part)
 
 
-def timename_from_datestr(given_date: str, part_count: int) -> str:
+def timename_from_datestr(given_date: str, timedepth: int) -> str:
     """
     >>> timename_from_datestr('2345-06-07', 1)
     '2345'
@@ -384,28 +374,32 @@ def timename_from_datestr(given_date: str, part_count: int) -> str:
     """
     return timename_from_date(
         datetime.datetime.fromisoformat(given_date),
-        part_count=part_count,
+        timedepth=timedepth,
     )
 
 
-def timename_from_date(given_date: datetime.date, part_count: int) -> str:
+def timename_from_date(given_date: datetime.date, timedepth: int) -> str:
     """
     >>> timename_from_date(datetime.date(3456, 7, 8), 2)
     '3456_07'
     """
-    _timeparts = itertools.islice(_each_timepart_from_date(given_date), part_count)
+    _timeparts = itertools.islice(_each_timepart_from_date(given_date), timedepth)
     return _format_timename(*_timeparts)
 
 
-def timeparts_from_date(given_date: datetime.date, part_count: int) -> tuple[int, ...]:
+def timeparts_from_date(given_date: datetime.date, timedepth: int) -> tuple[int, ...]:
     """
     >>> timeparts_from_date(datetime.date(3456, 7, 8), 2)
     (3456, 7)
     >>> timeparts_from_date(datetime.date(3456, 7, 8), 4)
     (3456, 7, 8, 0)
     """
-    _parts = itertools.chain(_each_timepart_from_date(given_date), itertools.repeat(0))
-    return tuple(itertools.islice(_parts, part_count))
+    return tuple(
+        itertools.islice(
+            _zeropadded_timeparts(_each_timepart_from_date(given_date)),
+            timedepth,
+        )
+    )
 
 
 def _each_timepart_from_date(given_date: datetime.date) -> Iterator[int]:
@@ -418,5 +412,66 @@ def _each_timepart_from_date(given_date: datetime.date) -> Iterator[int]:
         yield given_date.second
 
 
+def _zeropadded_timeparts(
+    timeparts: collections.abc.Iterable[int],
+) -> collections.abc.Iterator[int]:
+    yield from timeparts
+    yield from itertools.repeat(0)
+
+
 def _format_timepart(timepart: int) -> str:
     return f"{timepart:0{_TIMEPART_MIN_LEN}}"
+
+
+def _whenparts(
+    when: tuple[int, ...] | datetime.date,
+    timedepth: int,
+) -> tuple[int, ...]:
+    return (
+        timeparts_from_date(when, timedepth)
+        if isinstance(when, datetime.date)
+        else tuple(itertools.islice(_zeropadded_timeparts(when), timedepth))
+    )
+
+
+def full_semverlike_timeparts(when: tuple[int, ...] | datetime.date) -> str:
+    """
+    >>> full_semverlike_timeparts((3000, 2))
+    '3000.2'
+    >>> full_semverlike_timeparts(datetime.date(3000, 7, 12))
+    '3000.7.12'
+    >>> full_semverlike_timeparts(datetime.datetime(3000, 9, 1))
+    '3000.9.1.0.0.0'
+    """
+    _parts = (
+        timeparts_from_date(
+            when, timedepth=(6 if isinstance(when, datetime.datetime) else 3)
+        )
+        if isinstance(when, datetime.date)
+        else when
+    )
+    return ".".join(map(str, _parts))
+
+
+def semverlike_timeparts(when: tuple[int, ...] | datetime.date, timedepth: int) -> str:
+    """
+    >>> semverlike_timeparts((3000, 2, 7, 9), timedepth=2)
+    '3000.2'
+    >>> semverlike_timeparts((3000, 2), timedepth=1)
+    '3000'
+    >>> semverlike_timeparts((3000, 2, 7, 9), timedepth=5)
+    '3000.2.7.9.0'
+
+    >>> semverlike_timeparts(datetime.date(3000, 7, 12), timedepth=3)
+    '3000.7.12'
+    >>> semverlike_timeparts(datetime.date(3000, 7, 12), timedepth=2)
+    '3000.7'
+    >>> semverlike_timeparts(datetime.date(3000, 7, 12), timedepth=1)
+    '3000'
+
+    >>> semverlike_timeparts(datetime.datetime(3000, 9, 1, 5, 2), timedepth=6)
+    '3000.9.1.5.2.0'
+    >>> semverlike_timeparts(datetime.datetime(3000, 9, 1, 5, 2), timedepth=3)
+    '3000.9.1'
+    """
+    return full_semverlike_timeparts(_whenparts(when, timedepth))