Skip to content

Commit

Permalink
add parsing of pendulum_dt from unix time
Browse files Browse the repository at this point in the history
  • Loading branch information
07pepa committed Jun 11, 2024
1 parent 7097d98 commit 1e55159
Show file tree
Hide file tree
Showing 3 changed files with 150 additions and 2 deletions.
29 changes: 29 additions & 0 deletions pydantic_extra_types/pendulum_dt.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,25 @@
raise RuntimeError(
'The `pendulum_dt` module requires "pendulum" to be installed. You can install it with "pip install pendulum".'
) from e
import math
import re
from datetime import date, datetime, timedelta
from typing import Any, List, Type

from pydantic import GetCoreSchemaHandler
from pydantic_core import PydanticCustomError, core_schema

int_pattern = re.compile(r'^[+-]?\d+$')
int_as_float_pattern = re.compile(r'^[+-]?\d+.0+$') # for better precision when +-xxx.0
float_pattern = re.compile(r'^[+-]?\d*\.\d+$|^[+-]?inf$|^[+-]?nan$|^[+-]?infinity$', re.IGNORECASE)


def validate_float_datetime_timestamp(value: float) -> None:
if math.isinf(value):
raise ValueError('value is infinite and not a valid timestamp')
if math.isnan(value):
raise ValueError('value is not a number and not a valid timestamp')


class DateTime(_DateTime):
"""
Expand Down Expand Up @@ -68,6 +81,20 @@ def _validate(cls, value: Any, handler: core_schema.ValidatorFunctionWrapHandler
# if we are passed an existing instance, pass it straight through.
if isinstance(value, (_DateTime, datetime)):
return DateTime.instance(value)
if isinstance(value, int):
return DateTime.fromtimestamp(value)
if isinstance(value, float):
validate_float_datetime_timestamp(value)
return DateTime.fromtimestamp(value)
if isinstance(value, str):
if int_pattern.match(value):
return DateTime.fromtimestamp(int(value))
if int_as_float_pattern.match(value):
return DateTime.fromtimestamp(int(value.split('.', 1)[0]))
if float_pattern.match(value):
float_value = float(value)
validate_float_datetime_timestamp(float_value)
return DateTime.fromtimestamp(float_value)

# otherwise, parse it.
try:
Expand All @@ -84,6 +111,8 @@ def _validate(cls, value: Any, handler: core_schema.ValidatorFunctionWrapHandler
value.microsecond,
value.tzinfo,
)
except ValueError:
raise
except Exception as exc:
raise PydanticCustomError('value_error', 'value is not a valid timestamp') from exc

Expand Down
2 changes: 2 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,8 @@ filterwarnings = [
'error',
# This ignore will be removed when pycountry will drop py36 & support py311
'ignore:::pkg_resources',
# This ignore will be removed when pendulum fixes https://github.com/sdispater/pendulum/issues/834
'ignore:datetime.datetime.utcfromtimestamp.*:DeprecationWarning'
]

# configuring https://github.com/pydantic/hooky
Expand Down
121 changes: 119 additions & 2 deletions tests/test_pendulum_dt.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import re
from datetime import date, datetime, timedelta
from datetime import timezone as tz

Expand All @@ -10,6 +11,11 @@
UTC = tz.utc


int_pattern = re.compile(r'^[+-]?\d+$')
int_as_float_pattern = re.compile(r'^[+-]?\d+.0+$') # for beter precision when xxx.0
float_pattern = re.compile(r'^[+-]?\d*\.\d+$|^[+-]?inf$|^[+-]?nan$|^[+-]?infinity$', re.IGNORECASE)


class DtModel(BaseModel):
dt: DateTime

Expand Down Expand Up @@ -119,6 +125,49 @@ def test_pendulum_dt_from_serialized(dt):
assert isinstance(model.dt, pendulum.DateTime)


@pytest.mark.parametrize(
'dt',
[
pendulum.now().to_iso8601_string(),
pendulum.now().to_w3c_string(),
1718096578,
1718096578.5,
-5,
-5.5,
float('-0'),
'1718096578',
'1718096578.5',
'-5',
'-5.5',
'-0',
'-0.0',
'+0.0',
'+1718096578.5',
],
)
def test_pendulum_dt_from_str_unix_timestamp(dt):
"""
Verifies that building an instance from serialized, well-formed strings decode properly.
"""
if isinstance(dt, str):
if int_pattern.match(dt):
dt_actual = DateTime.fromtimestamp(int(dt))
elif int_as_float_pattern.match(dt):
dt_actual = DateTime.fromtimestamp(int(dt.split('.', 1)[0]))
elif float_pattern.match(dt):
dt_actual = DateTime.fromtimestamp(float(dt))
else:
dt_actual = pendulum.parse(dt)
elif isinstance(dt, int | float):
dt_actual = pendulum.from_timestamp(dt)
else:
dt_actual = pendulum.parse(dt)
model = DtModel(dt=dt)
assert model.dt == dt_actual
assert type(model.dt) is DateTime
assert isinstance(model.dt, pendulum.DateTime)


@pytest.mark.parametrize(
'd',
[pendulum.now().date().isoformat(), pendulum.now().to_w3c_string(), pendulum.now().to_iso8601_string()],
Expand Down Expand Up @@ -155,7 +204,15 @@ def test_pendulum_duration_from_serialized(delta_t_str):
assert isinstance(model.delta_t, pendulum.Duration)


@pytest.mark.parametrize('dt', [None, 'malformed', pendulum.now().to_iso8601_string()[:5], 42, 'P10Y10M10D'])
@pytest.mark.parametrize(
'dt',
[
None,
'malformed',
pendulum.now().to_iso8601_string()[:5],
'P10Y10M10D',
],
)
def test_pendulum_dt_malformed(dt):
"""
Verifies that the instance fails to validate if malformed dt are passed.
Expand All @@ -164,7 +221,67 @@ def test_pendulum_dt_malformed(dt):
DtModel(dt=dt)


@pytest.mark.parametrize('date', [None, 'malformed', pendulum.today().to_iso8601_string()[:5], 42, 'P10Y10M10D'])
@pytest.mark.parametrize(
'dt',
[
float('inf'),
float('-inf'),
'inf',
'-inf',
'INF',
'-INF',
'+inf',
'Infinity',
'+Infinity',
'-Infinity',
'INFINITY',
'+INFINITY',
'-INFINITY',
'infinity',
'+infinity',
'-infinity',
],
)
def test_pendulum_dt_infinite(dt):
"""
Verifies that the instance fails to validate if malformed dt are passed.
"""
with pytest.raises(
ValidationError,
match=re.escape(
'1 validation error for DtModel\ndt\n ' 'Value error, value is infinite and not a valid timestamp'
)
+ '.*',
):
DtModel(dt=dt)


@pytest.mark.parametrize(
'dt',
[
float('nan'),
'nan',
'NaN',
'NAN',
'+nan',
'-nan',
],
)
def test_pendulum_dt_nan(dt):
"""
Verifies that the instance fails to validate if malformed dt are passed.
"""
with pytest.raises(
ValidationError,
match=re.escape(
'1 validation error for DtModel\ndt\n ' 'Value error, value is not a number' ' and not a valid timestamp'
)
+ '.*',
):
DtModel(dt=dt)


@pytest.mark.parametrize('date', [None, 'malformed', pendulum.today().to_iso8601_string()[:5], 'P10Y10M10D'])
def test_pendulum_date_malformed(date):
"""
Verifies that the instance fails to validate if malformed date are passed.
Expand Down

0 comments on commit 1e55159

Please sign in to comment.