Merge pull request #233 from steersbob/feature/timedelta-parse

feature/timedelta parse

brewblox_history/models.py

@@ -4,11 +4,44 @@
 
 import collections
 from datetime import datetime, timedelta
-from typing import Any, Literal, NamedTuple
+from typing import Annotated, Any, Literal, NamedTuple
 
 from pydantic import (BaseModel, ConfigDict, Field, field_validator,
                       model_validator)
+from pydantic.functional_validators import BeforeValidator
+from pydantic_core import SchemaValidator, core_schema
 from pydantic_settings import BaseSettings, SettingsConfigDict
+from pytimeparse.timeparse import timeparse
+
+DurationSrc_ = str | int | float | timedelta
+DatetimeSrc_ = str | int | float | datetime | None
+
+pydantic_timedelta_validator = SchemaValidator(core_schema.timedelta_schema())
+pydantic_datetime_validator = SchemaValidator(core_schema.datetime_schema())
+
+
+def parse_duration(value: DurationSrc_) -> timedelta:
+    if isinstance(value, timedelta):
+        return value
+
+    try:
+        value = float(value)
+    except TypeError:
+        value = None
+    except ValueError:
+        value = timeparse(value) or value
+
+    return pydantic_timedelta_validator.validate_python(value)
+
+
+def parse_datetime(value: DatetimeSrc_) -> datetime | None:
+    if value is None or value == '':
+        return None
+
+    return pydantic_datetime_validator.validate_python(value)
+
+
+loose_timedelta = Annotated[timedelta, BeforeValidator(parse_duration)]
 
 
 def flatten(d, parent_key=''):
@@ -57,11 +90,11 @@ class ServiceConfig(BaseSettings):
     history_topic: str = 'brewcast/history'
     datastore_topic: str = 'brewcast/datastore'
 
-    ranges_interval: timedelta = timedelta(seconds=10)
-    metrics_interval: timedelta = timedelta(seconds=10)
-    minimum_step: timedelta = timedelta(seconds=10)
+    ranges_interval: loose_timedelta = timedelta(seconds=10)
+    metrics_interval: loose_timedelta = timedelta(seconds=10)
+    minimum_step: loose_timedelta = timedelta(seconds=10)
 
-    query_duration_default: timedelta = timedelta(days=1)
+    query_duration_default: loose_timedelta = timedelta(days=1)
     query_desired_points: int = 1000
 
 
@@ -117,12 +150,14 @@ class DatastoreDeleteResponse(BaseModel):
 
 
 class TimeSeriesFieldsQuery(BaseModel):
-    duration: str = Field('1d', examples=['10m', '1d'])
+    duration: loose_timedelta = Field(timedelta(days=1),
+                                      examples=['10m', '1d'])
 
 
 class TimeSeriesMetricsQuery(BaseModel):
     fields: list[str]
-    duration: str = Field('10m', examples=['10m', '1d'])
+    duration: loose_timedelta = Field(timedelta(minutes=10),
+                                      examples=['10m', '1d'])
 
 
 class TimeSeriesMetric(BaseModel):
@@ -135,7 +170,7 @@ class TimeSeriesRangesQuery(BaseModel):
     fields: list[str] = Field(examples=[['spark-one/sensor/value[degC]']])
     start: datetime | None = Field(None, examples=['2020-01-01T20:00:00.000Z'])
     end: datetime | None = Field(None, examples=['2030-01-01T20:00:00.000Z'])
-    duration: str | None = Field(None, examples=['1d'])
+    duration: loose_timedelta | None = Field(None, examples=['1d'])
 
 
 class TimeSeriesRangeValue(NamedTuple):

brewblox_history/utils.py

@@ -3,16 +3,11 @@
 from datetime import datetime, timedelta, timezone
 from functools import lru_cache
 
-import ciso8601
-from pytimeparse.timeparse import timeparse
-
-from .models import ServiceConfig
+from .models import (DatetimeSrc_, DurationSrc_, ServiceConfig, parse_datetime,
+                     parse_duration)
 
 LOGGER = logging.getLogger(__name__)
 
-DurationSrc_ = str | int | float | timedelta
-DatetimeSrc_ = str | int | float | datetime | None
-
 
 class DuplicateFilter(logging.Filter):
     """
@@ -47,40 +42,6 @@ def strex(ex: Exception, tb=False):
         return msg
 
 
-def parse_duration(value: DurationSrc_) -> timedelta:
-    if isinstance(value, timedelta):
-        return value
-
-    try:
-        total_seconds = float(value)
-    except ValueError:
-        total_seconds = timeparse(value)
-
-    return timedelta(seconds=total_seconds)
-
-
-def parse_datetime(value: DatetimeSrc_) -> datetime | None:
-    if value is None or value == '':
-        return None
-
-    elif isinstance(value, datetime):
-        return value
-
-    elif isinstance(value, str):
-        return ciso8601.parse_datetime(value)
-
-    elif isinstance(value, (int, float)):
-        # This is an educated guess
-        # 10e10 falls in 1973 if the timestamp is in milliseconds,
-        # and in 5138 if the timestamp is in seconds
-        if value > 10e10:
-            value /= 1000
-        return datetime.fromtimestamp(value, tz=timezone.utc)
-
-    else:
-        raise ValueError(str(value))
-
-
 def format_datetime(value: DatetimeSrc_, precision: str = 's') -> str:
     """Formats given date/time value with desired precision.
 

brewblox_history/victoria.py

@@ -51,7 +51,7 @@ async def _json_query(self, query: str, url: str):
         return resp.json()
 
     async def fields(self, args: TimeSeriesFieldsQuery) -> list[str]:
-        query = f'match[]={{__name__!=""}}&start={args.duration}'
+        query = f'match[]={{__name__!=""}}&start={args.duration.total_seconds()}s'
         LOGGER.debug(query)
         result = await self._json_query(query, '/api/v1/series')
         retv = [

test/test_utils.py

@@ -6,6 +6,7 @@
 from datetime import datetime, timedelta, timezone
 
 import pytest
+from pydantic import ValidationError
 
 from brewblox_history import utils
 
@@ -46,11 +47,12 @@ def test_parse_duration():
     assert utils.parse_duration('2h10m') == timedelta(hours=2, minutes=10)
     assert utils.parse_duration('10') == timedelta(seconds=10)
     assert utils.parse_duration(timedelta(hours=1)) == timedelta(minutes=60)
+    assert utils.parse_duration('P1DT10M5S') == timedelta(days=1, minutes=10, seconds=5)
 
-    with pytest.raises(TypeError):
+    with pytest.raises(ValidationError):
         utils.parse_duration('')
 
-    with pytest.raises(TypeError):
+    with pytest.raises(ValidationError):
         utils.parse_duration(None)