-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add unicode datetime attribute (#33)
UnicodeDatetimeAttribute stores datetimes as 8601 ISO strings with offset. The storage representation of this format will look something like: {"key": {"S": "2020-11-22T03:22:33.444444-08:00"}} The attribute by default will add an offset to UTC if not present and make it timezone aware. It also as options for normalizing the date to UTC (for caching purposes) and adds support for custom formatting.
- Loading branch information
Showing
6 changed files
with
293 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
from datetime import datetime | ||
from datetime import timezone | ||
from typing import Any | ||
from typing import Optional | ||
|
||
import pynamodb.attributes | ||
from pynamodb.attributes import Attribute | ||
|
||
|
||
class UnicodeDatetimeAttribute(Attribute[datetime]): | ||
""" | ||
Stores a 'datetime.datetime' object as an ISO8601 formatted string | ||
This is useful for wanting database readable datetime objects that also sort. | ||
>>> class MyModel(Model): | ||
>>> created_at = UnicodeDatetimeAttribute() | ||
""" | ||
|
||
attr_type = pynamodb.attributes.STRING | ||
|
||
def __init__( | ||
self, | ||
*, | ||
force_tz: bool = True, | ||
force_utc: bool = False, | ||
fmt: Optional[str] = None, | ||
**kwargs: Any, | ||
) -> None: | ||
""" | ||
:param force_tz: If set it will add timezone info to the `datetime` value if no `tzinfo` is currently | ||
set before serializing, defaults to `True` | ||
:param force_utc: If set it will normalize the `datetime` to UTC before serializing the value | ||
:param fmt: If set this value will be used to format the `datetime` object for serialization | ||
and deserialization | ||
""" | ||
|
||
super().__init__(**kwargs) | ||
self._force_tz = force_tz | ||
self._force_utc = force_utc | ||
self._fmt = fmt | ||
|
||
def deserialize(self, value: str) -> datetime: | ||
return ( | ||
datetime.fromisoformat(value) | ||
if self._fmt is None | ||
else datetime.strptime(value, self._fmt) | ||
) | ||
|
||
def serialize(self, value: datetime) -> str: | ||
if self._force_tz and value.tzinfo is None: | ||
value = value.replace(tzinfo=timezone.utc) | ||
if self._force_utc: | ||
value = value.astimezone(tz=timezone.utc) | ||
return value.isoformat() if self._fmt is None else value.strftime(self._fmt) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,206 @@ | ||
from datetime import datetime | ||
from unittest.mock import ANY | ||
|
||
import pytest | ||
from pynamodb.attributes import UnicodeAttribute | ||
from pynamodb.models import Model | ||
|
||
from pynamodb_attributes import UnicodeDatetimeAttribute | ||
from tests.connection import _connection | ||
from tests.meta import dynamodb_table_meta | ||
|
||
|
||
CUSTOM_FORMAT = "%m/%d/%Y, %H:%M:%S" | ||
CUSTOM_FORMAT_DATE = "11/22/2020, 11:22:33" | ||
TEST_ISO_DATE_NO_OFFSET = "2020-11-22T11:22:33.444444" | ||
TEST_ISO_DATE_UTC = "2020-11-22T11:22:33.444444+00:00" | ||
TEST_ISO_DATE_PST = "2020-11-22T03:22:33.444444-08:00" | ||
|
||
|
||
class MyModel(Model): | ||
Meta = dynamodb_table_meta(__name__) | ||
|
||
key = UnicodeAttribute(hash_key=True) | ||
default = UnicodeDatetimeAttribute(null=True) | ||
no_force_tz = UnicodeDatetimeAttribute(force_tz=False, null=True) | ||
force_utc = UnicodeDatetimeAttribute(force_utc=True, null=True) | ||
force_utc_no_force_tz = UnicodeDatetimeAttribute( | ||
force_utc=True, | ||
force_tz=False, | ||
null=True, | ||
) | ||
custom_format = UnicodeDatetimeAttribute(fmt=CUSTOM_FORMAT, null=True) | ||
|
||
|
||
@pytest.fixture(scope="module", autouse=True) | ||
def create_table(): | ||
MyModel.create_table() | ||
|
||
|
||
@pytest.mark.parametrize( | ||
["value", "expected_str", "expected_value"], | ||
[ | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_NO_OFFSET), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
TEST_ISO_DATE_PST, | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
), | ||
], | ||
) | ||
def test_default_serialization(value, expected_str, expected_value, uuid_key): | ||
model = MyModel() | ||
model.key = uuid_key | ||
model.default = value | ||
|
||
model.save() | ||
|
||
actual = MyModel.get(hash_key=uuid_key) | ||
assert actual.default == expected_value | ||
|
||
item = _connection(MyModel).get_item(uuid_key) | ||
assert item["Item"] == {"key": ANY, "default": {"S": expected_str}} | ||
|
||
|
||
@pytest.mark.parametrize( | ||
["value", "expected_str", "expected_value"], | ||
[ | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_NO_OFFSET), | ||
TEST_ISO_DATE_NO_OFFSET, | ||
datetime.fromisoformat(TEST_ISO_DATE_NO_OFFSET), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
TEST_ISO_DATE_PST, | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
), | ||
], | ||
) | ||
def test_no_force_tz_serialization(value, expected_str, expected_value, uuid_key): | ||
model = MyModel() | ||
model.key = uuid_key | ||
model.no_force_tz = value | ||
|
||
model.save() | ||
|
||
actual = MyModel.get(hash_key=uuid_key) | ||
item = _connection(MyModel).get_item(uuid_key) | ||
|
||
assert item["Item"] == {"key": ANY, "no_force_tz": {"S": expected_str}} | ||
|
||
assert actual.no_force_tz == expected_value | ||
|
||
|
||
@pytest.mark.parametrize( | ||
["value", "expected_str", "expected_value"], | ||
[ | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_NO_OFFSET), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
], | ||
) | ||
def test_force_utc_serialization(value, expected_str, expected_value, uuid_key): | ||
model = MyModel() | ||
model.key = uuid_key | ||
model.force_utc = value | ||
|
||
model.save() | ||
|
||
actual = MyModel.get(hash_key=uuid_key) | ||
item = _connection(MyModel).get_item(uuid_key) | ||
|
||
assert item["Item"] == {"key": ANY, "force_utc": {"S": expected_str}} | ||
|
||
assert actual.force_utc == expected_value | ||
|
||
|
||
@pytest.mark.parametrize( | ||
["value", "expected_str", "expected_value"], | ||
[ | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_PST), | ||
TEST_ISO_DATE_UTC, | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
), | ||
], | ||
) | ||
def test_force_utc_no_force_tz_serialization( | ||
value, | ||
expected_str, | ||
expected_value, | ||
uuid_key, | ||
): | ||
model = MyModel() | ||
model.key = uuid_key | ||
model.force_utc_no_force_tz = value | ||
|
||
model.save() | ||
|
||
actual = MyModel.get(hash_key=uuid_key) | ||
item = _connection(MyModel).get_item(uuid_key) | ||
|
||
assert item["Item"] == {"key": ANY, "force_utc_no_force_tz": {"S": expected_str}} | ||
|
||
assert actual.force_utc_no_force_tz == expected_value | ||
|
||
|
||
@pytest.mark.parametrize( | ||
["value", "expected_str", "expected_value"], | ||
[ | ||
( | ||
datetime.fromisoformat(TEST_ISO_DATE_UTC), | ||
CUSTOM_FORMAT_DATE, | ||
datetime(2020, 11, 22, 11, 22, 33), | ||
), | ||
], | ||
) | ||
def test_custom_format_force_tz_serialization( | ||
value, | ||
expected_str, | ||
expected_value, | ||
uuid_key, | ||
): | ||
model = MyModel() | ||
model.key = uuid_key | ||
model.custom_format = value | ||
|
||
model.save() | ||
|
||
actual = MyModel.get(hash_key=uuid_key) | ||
item = _connection(MyModel).get_item(uuid_key) | ||
|
||
assert item["Item"] == {"key": ANY, "custom_format": {"S": expected_str}} | ||
|
||
assert actual.custom_format == expected_value |