import base64
import collections.abc
import datetime
import decimal
import numbers
import uuid
from typing import ClassVar
from . import actions
ENCODING = "utf-8"
STRING = "S"
NUMBER = "N"
BINARY = "B"
BOOLEAN = "BOOL"
MAP = "M"
LIST = "L"
PRIMITIVES = {"S", "N", "B"}
SETS = {"SS", "NS", "BS"}
DOCUMENTS = {"L", "M"}
ALL = {*PRIMITIVES, *SETS, *DOCUMENTS, BOOLEAN}
# Dynamo takes numbers as strings to reduce inter-language problems
DYNAMODB_CONTEXT = decimal.Context(
Emin=-128, Emax=126, rounding=None, prec=38,
traps=[
decimal.Clamped, decimal.Overflow, decimal.Inexact,
decimal.Rounded, decimal.Underflow
]
)
OPERATION_SUPPORT_BY_OP = {
"==": ALL,
"!=": ALL,
"<": PRIMITIVES,
">": PRIMITIVES,
"<=": PRIMITIVES,
">=": PRIMITIVES,
"begins_with": {STRING, BINARY},
"between": PRIMITIVES,
"contains": {*SETS, STRING, BINARY, LIST},
"in": ALL
}
OPERATION_SUPPORT_BY_TYPE = {
t: {op for (op, supported) in OPERATION_SUPPORT_BY_OP.items() if t in supported}
for t in ALL
}
[docs]class Type:
"""Abstract base type."""
python_type = None
backing_type = None
[docs] def supports_operation(self, operation: str) -> bool:
"""
Used to ensure a conditional operation is supported by this type.
By default, uses a hardcoded table of operations that maps to each backing DynamoDB type.
You can override this method to implement your own conditional operators, or to dynamically
adjust which operations your type supports.
"""
return operation in OPERATION_SUPPORT_BY_TYPE[self.backing_type]
def __init__(self):
if not hasattr(self, "inner_typedef"):
self.inner_typedef = self
super().__init__()
def __getitem__(self, key):
raise RuntimeError(f"{self!r} does not support document paths")
[docs] def dynamo_dump(self, value, *, context, **kwargs):
"""Converts a local value into a DynamoDB value.
For example, to store a string enum as an integer:
.. code-block:: python
def dynamo_dump(self, value, *, context, **kwargs):
colors = ["red", "blue", "green"]
return colors.index(value.lower())
"""
raise NotImplementedError
[docs] def dynamo_load(self, value, *, context, **kwargs):
"""Converts a DynamoDB value into a local value.
For example, to load a string enum from an integer:
.. code-block:: python
def dynamo_dump(self, value, *, context, **kwargs):
colors = ["red", "blue", "green"]
return colors[value]
"""
raise NotImplementedError
[docs] def _dump(self, value, **kwargs):
"""Entry point for serializing values. Most custom types should use :func:`~bloop.types.Type.dynamo_dump`.
This wraps the return value of :func:`~bloop.types.Type.dynamo_dump` in DynamoDB's wire format.
For example, serializing a string enum to an int:
.. code-block:: python
value = "green"
# dynamo_dump("green") = 2
_dump(value) == {"N": 2}
If a complex type calls this function with ``None``, it will forward ``None`` to
:func:`~bloop.types.Type.dynamo_dump`. This can happen when dumping eg. a sparse
:class:`~.bloop.types.Map`, or a missing (not set) value.
"""
wrapped = actions.wrap(value)
value = self.dynamo_dump(wrapped.value, **kwargs)
if value is None:
return actions.wrap(None)
else:
value = {self.backing_type: value}
return wrapped.type.new_action(value)
[docs] def _load(self, value, **kwargs):
"""Entry point for deserializing values. Most custom types should use :func:`~bloop.types.Type.dynamo_load`.
This unpacks DynamoDB's wire format and calls :func:`~bloop.types.Type.dynamo_load` on the inner value.
For example, deserializing an int to a string enum:
.. code-block:: python
value = {"N": 2}
# dynamo_load(2) = "green"
_load(value) == "green"
If a complex type calls this function with ``None``, it will forward ``None`` to
:func:`~bloop.types.Type.dynamo_load`. This can happen when loading eg. a sparse :class:`~bloop.types.Map`.
"""
if value is not None:
value = next(iter(value.values()))
return self.dynamo_load(value, **kwargs)
def __repr__(self):
# Render class python types by name
python_type = self.python_type
if isinstance(python_type, type):
python_type = python_type.__name__
return "<{}[{}:{}]>".format(
self.__class__.__name__,
self.backing_type, python_type
)
[docs]class String(Type):
python_type = str
backing_type = STRING
def dynamo_load(self, value, *, context, **kwargs):
if not value:
return ""
return value
def dynamo_dump(self, value, *, context, **kwargs):
if not value:
return None
return value
[docs]class UUID(String):
python_type = uuid.UUID
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
return uuid.UUID(value)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
return str(value)
FIXED_ISO8601_FORMAT = "%Y-%m-%dT%H:%M:%S.%f+00:00"
[docs]class DateTime(String):
"""Always stored in DynamoDB using the :data:`~bloop.types.FIXED_ISO8601_FORMAT` format.
Naive datetimes (``tzinfo is None``) are not supported, and trying to use one will raise ``ValueError``.
.. code-block:: python
from datetime import datetime, timedelta, timezone
class Model(Base):
id = Column(Integer, hash_key=True)
date = Column(DateTime)
engine.bind()
obj = Model(id=1, date=datetime.now(timezone.utc))
engine.save(obj)
one_day_ago = datetime.now(timezone.utc) - timedelta(days=1)
query = engine.query(
Model,
key=Model.id==1,
filter=Model.date >= one_day_ago)
query.first().date
.. note::
To use common datetime libraries such as `arrow`_, `delorean`_, or `pendulum`_,
see :ref:`DateTime and Timestamp Extensions <user-extensions-datetime>` in the user guide. These
are drop-in replacements and support non-utc timezones:
.. code-block:: python
from bloop import DateTime # becomes:
from bloop.ext.pendulum import DateTime
.. _arrow: http://crsmithdev.com/arrow
.. _delorean: https://delorean.readthedocs.io/en/latest/
.. _pendulum: https://pendulum.eustace.io
"""
python_type = datetime.datetime
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
dt = datetime.datetime.strptime(value, FIXED_ISO8601_FORMAT)
# we assume all stored values are utc, so we simply force timezone to utc
# without changing the day/time values
return dt.replace(tzinfo=datetime.timezone.utc)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
if value.tzinfo is None:
raise ValueError(
"naive datetime instances are not supported. You can set a timezone with either "
"your_dt.replace(tzinfo=) or your_dt.astimezone(tz=). WARNING: calling astimezone on a naive "
"datetime will assume the naive datetime is in the system's timezone, even though "
"datetime.utcnow() creates a naive object! You almost certainly don't want to do that."
)
dt = value.astimezone(tz=datetime.timezone.utc)
return dt.strftime(FIXED_ISO8601_FORMAT)
[docs]class Number(Type):
"""Base for all numeric types.
:param context: *(Optional)* :class:`decimal.Context` used to translate numbers. Default is a context that
matches DynamoDB's `stated limits`__, taken from `boto3`__.
__ https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Limits.html#limits-data-types-numbers
__ https://github.com/boto/boto3/blob/dffeb393a795204f375b951d791c768be6b1cb8c/boto3/dynamodb/types.py#L32
"""
python_type = decimal.Decimal
backing_type = NUMBER
def __init__(self, context=None):
self.context = context or DYNAMODB_CONTEXT
super().__init__()
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
return self.context.create_decimal(value)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
n = str(self.context.create_decimal(value))
if any(filter(lambda x: x in n, ("Infinity", "NaN"))):
raise TypeError("{!r} does not support Infinity and NaN.".format(self))
return n
[docs]class Integer(Number):
"""Truncates values when loading or dumping.
For example, ``3.14`` in DynamoDB is loaded as ``3``. If a value is ``7.5``
locally, it's stored in DynamoDB as ``7``.
"""
python_type = int
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
number = super().dynamo_load(value, context=context, **kwargs)
return int(number)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
value = int(value)
return super().dynamo_dump(value, context=context, **kwargs)
[docs]class Timestamp(Integer):
"""Stores the unix (epoch) time in seconds. Milliseconds are truncated to 0 on load and save.
Naive datetimes (``tzinfo is None``) are not supported, and trying to use one will raise ``ValueError``.
.. code-block:: python
from datetime import datetime, timedelta, timezone
class Model(Base):
id = Column(Integer, hash_key=True)
date = Column(Timestamp)
engine.bind()
obj = Model(id=1, date=datetime.now(timezone.utc))
engine.save(obj)
one_day_ago = datetime.now(timezone.utc) - timedelta(days=1)
query = engine.query(
Model,
key=Model.id==1,
filter=Model.date >= one_day_ago)
query.first().date
.. note::
To use common datetime libraries such as `arrow`_, `delorean`_, or `pendulum`_,
see :ref:`DateTime and Timestamp Extensions <user-extensions-datetime>` in the user guide. These
are drop-in replacements and support non-utc timezones:
.. code-block:: python
from bloop import Timestamp # becomes:
from bloop.ext.pendulum import Timestamp
.. _arrow: http://crsmithdev.com/arrow
.. _delorean: https://delorean.readthedocs.io/en/latest/
.. _pendulum: https://pendulum.eustace.io
"""
python_type = datetime.datetime
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
value = super().dynamo_load(value, context=context, **kwargs)
# we assume all stored values are utc, so we simply force timezone to utc
# without changing the day/time values
return datetime.datetime.fromtimestamp(value, tz=datetime.timezone.utc)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
if value.tzinfo is None:
raise ValueError(
"naive datetime instances are not supported. You can set a timezone with either "
"your_dt.replace(tzinfo=) or your_dt.astimezone(tz=). WARNING: calling astimezone on a naive "
"datetime will assume the naive datetime is in the system's timezone, even though "
"datetime.utcnow() creates a naive object! You almost certainly don't want to do that."
)
value = value.timestamp()
return super().dynamo_dump(value, context=context, **kwargs)
[docs]class Binary(Type):
python_type = bytes
backing_type = BINARY
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return b""
return base64.b64decode(value)
def dynamo_dump(self, value, *, context, **kwargs):
if not value:
return None
return base64.b64encode(value).decode("utf-8")
[docs]class Boolean(Type):
python_type = bool
backing_type = BOOLEAN
def dynamo_load(self, value, *, context, **kwargs):
if value is None:
return None
return bool(value)
def dynamo_dump(self, value, *, context, **kwargs):
if value is None:
return None
return bool(value)
def subclassof(c, b):
"""Wrap issubclass to return True/False without throwing TypeError"""
try:
return issubclass(c, b)
except TypeError:
return False
def type_instance(typedef):
"""Returns an instance of a type class, or the instance if provided"""
if subclassof(typedef, Type):
# Type class passed, create no-arg instance
typedef = typedef()
return typedef
def guard_no_action(func):
def call(value, *, context, **kwargs):
# guard call to _load or _dump
if isinstance(value, actions.Action):
if not value.type.nestable:
raise ValueError(f"cannot nest the action type {value.type}")
value = actions.unwrap(value)
value = func(value, context=context, **kwargs)
# guard response from _dump
if isinstance(value, actions.Action):
if not value.type.nestable:
raise ValueError(f"cannot nest the action type {value.type}")
value = actions.unwrap(value)
return value
return call
[docs]class Set(Type):
"""Generic set type. Must provide an inner type.
.. code-block:: python
class Customer(BaseModel):
id = Column(Integer, hash_key=True)
account_ids = Column(Set(UUID))
:param typedef: The type to use when loading and saving values in this set.
Must have a ``backing_type`` of "S", "N", or "B".
"""
python_type = collections.abc.Set
def __init__(self, typedef):
self.inner_typedef = type_instance(typedef)
self.backing_type = typedef.backing_type + "S"
if self.backing_type not in {"NS", "SS", "BS"}:
raise TypeError("{!r} is not a valid set type.".format(self.backing_type))
super().__init__()
def dynamo_load(self, values, *, context, **kwargs):
if values is None:
return set()
return set(
self.inner_typedef.dynamo_load(value, context=context, **kwargs)
for value in values)
def dynamo_dump(self, values, *, context, **kwargs):
if values is None:
return None
dumped = []
dump = guard_no_action(self.inner_typedef.dynamo_dump)
for value in values:
value = dump(value, context=context, **kwargs)
if value is not None:
dumped.append(value)
return dumped or None
[docs]class List(Type):
"""Holds values of a single type.
Similar to :class:`~bloop.types.Set` because it requires a single type. However, that type
can be another List, or :class:`~bloop.types.Map`, or :class:`~bloop.types.Boolean`. This is restricted
to a single type even though DynamoDB is not because there is no way to know which Type to load a DynamoDB value
with.
For example, ``{"S": "6d8b54a2-fa07-47e1-9305-717699459293"}`` could be loaded with
:class:`~bloop.types.UUID`, :class:`~bloop.types.String`, or any other class that is backed by "S".
.. code-block:: python
SingleQuizAnswers = List(String)
class AnswerBook(BaseModel):
...
all_answers = Column(List(SingleQuizAnswers))
.. seealso::
To store arbitrary lists, see :class:`~bloop.types.DynamicList`.
:param typedef: The type to use when loading and saving values in this list.
"""
python_type = collections.abc.Iterable
backing_type = LIST
def __init__(self, typedef):
self.inner_typedef = type_instance(typedef)
super().__init__()
def __getitem__(self, key):
return self.inner_typedef
def dynamo_load(self, values, *, context, **kwargs):
if values is None:
return list()
# noinspection PyProtectedMember
load = self.inner_typedef._load
return [
load(value, context=context, **kwargs)
for value in values]
def dynamo_dump(self, values, *, context, **kwargs):
if values is None:
return None
# noinspection PyProtectedMember
dump = guard_no_action(self.inner_typedef._dump)
dumped = (dump(value, context=context, **kwargs) for value in values)
return [value for value in dumped if value is not None] or None
[docs]class Map(Type):
"""Mapping of fixed keys and their Types.
.. code-block:: python
Metadata = Map(**{
"created": DateTime,
"referrer": UUID,
"cache": String
})
Product = Map(
id=Integer,
metadata=Metadata,
price=Number
)
class ProductCatalog(BaseModel):
...
all_products = Column(List(Product))
.. seealso::
To store arbitrary documents, see :class:`~bloop.types.DynamicMap`.
:param types: *(Optional)* specifies the keys and their Types when loading and dumping the Map.
Any keys that aren't specified in ``types`` are ignored when loading and dumping.
"""
python_type = collections.abc.Mapping
backing_type = MAP
def __init__(self, **types):
self.types = {k: type_instance(t) for k, t in types.items()}
super().__init__()
def __getitem__(self, key):
"""Overload allows easy nested access to types"""
return self.types[key]
def dynamo_load(self, values, *, context, **kwargs):
if values is None:
values = dict()
loaded = {}
for key, typedef in self.types.items():
# noinspection PyProtectedMember
value = typedef._load(values.get(key, None), context=context, **kwargs)
loaded[key] = value
return loaded
def dynamo_dump(self, values, *, context, **kwargs):
if values is None:
return None
dumped = {}
for key, typedef in self.types.items():
# noinspection PyProtectedMember
dump = guard_no_action(typedef._dump)
value = dump(values.get(key, None), context=context, **kwargs)
if value is not None:
dumped[key] = value
return dumped or None
[docs]class DynamicType(Type):
"""
Dynamically dumps a value based on its python type.
This is used by DynamicList, DynamicMap to handle path resolution before the value for an arbitrary path is known.
For example, given the following model:
.. code-block:: python
class UserUpload(BaseModel):
id = Column(String, hash_key=True)
doc = Column(DynamicMap)
And an instance as follows:
.. code-block:: python
u = UserUpload(id="numberoverzero")
u.doc = {
"foo": ["bar", {0: "a", 1: "b"}, True]
}
The renderer must know a type for ``UserUpload.doc["foo"][1][0]`` before the value is provided.
An instance of this type will return itself for any value during ``__getitem__``, and then inspects the value type
during _dump to create the correct simple type.
Because ``DynamicType`` requires access to the DynamoDB type annotation, you must call ``_load`` and ``_dump``,
as ``dynamo_load`` and ``dynamo_dump`` can't be implemented. For example:
.. code-block:: python
DynamicType.i._load({"S": "2016-08-09T01:16:25.322849+00:00"})
-> "2016-08-09T01:16:25.322849+00:00"
DynamicType.i._load({"N": "3.14"}) -> Decimal('3.14')
DynamicType.i._dump([1, True, "f"])
-> {"L": [{"N": "1"}, {"BOOL": true}, {"S": "f"}]}
DynamicType.i._dump({b"1", b"2"}) -> {"BS": ["MQ==", b"Mg=="]}
"""
i: ClassVar["DynamicType"]
def supports_operation(self, operation: str) -> bool:
"""Always True, because the type of the value passing through DynamicType is late bound"""
return True
def __getitem__(self, key):
"""Overload allows easy nested access to types"""
return self
def _load(self, value, **kwargs):
if value is None:
return None
vtype = DynamicType.extract_backing_type(value)
return DYNAMIC_TYPES[vtype]._load(value, **kwargs)
def _dump(self, value, **kwargs):
wrapped = actions.wrap(value)
if wrapped.value is None:
return wrapped
vtype = DynamicType.backing_type_for(wrapped.value)
return DYNAMIC_TYPES[vtype]._dump(wrapped, **kwargs)
def dynamo_load(self, value, *, context, **kwargs):
raise NotImplementedError
def dynamo_dump(self, value, *, context, **kwargs):
raise NotImplementedError
@staticmethod
def extract_backing_type(value: dict) -> str:
"""Returns the DynamoDB backing type from a given wire dict
::
{'S': 'foo'} -> 'S'
"""
return next(iter(value.keys()))
@staticmethod
def backing_type_for(value):
"""Returns the DynamoDB backing type for a given python value's type
::
4 -> 'N'
['x', 3] -> 'L'
{2, 4} -> 'SS'
"""
if isinstance(value, str):
vtype = "S"
elif isinstance(value, bytes):
vtype = "B"
# NOTE: numbers.Number check must come **AFTER** bool check since isinstance(True, numbers.Number)
elif isinstance(value, bool):
vtype = "BOOL"
elif isinstance(value, numbers.Number):
vtype = "N"
elif isinstance(value, dict):
vtype = "M"
elif isinstance(value, list):
vtype = "L"
elif isinstance(value, set):
if not value:
vtype = "SS" # doesn't matter, Set(x) should dump an empty set the same for all x
else:
inner = next(iter(value))
if isinstance(inner, str):
vtype = "SS"
elif isinstance(inner, bytes):
vtype = "BS"
elif isinstance(inner, numbers.Number):
vtype = "NS"
else:
raise ValueError(f"Unknown set type for inner value {inner!r}")
else:
raise ValueError(f"Can't dump unexpected type {type(value)!r} for value {value!r}")
return vtype
# Singleton instance for re-use.
# It's unlikely we'll ever need more than one.
DynamicType.i = DynamicType()
[docs]class DynamicList(Type):
"""Holds a list of arbitrary values, including other DynamicLists and DynamicMaps.
Similar to :class:`~bloop.types.List` but is not constrained to a single type.
.. code-block:: python
value = [1, True, "f"]
DynamicList()._dump(value)
-> {"L": [{"N": "1"}, {"BOOL": true}, {"S": "f"}]}
.. note::
Values will only be loaded and dumped as their DynamoDB backing types. This means datetimes and uuids are
stored and loaded as strings, and timestamps are stored and loaded as integers. For more information, see
:ref:`dynamic-documents`.
"""
python_type = collections.abc.Iterable
backing_type = LIST
def __getitem__(self, key):
"""Overload allows easy nested access to types"""
return DynamicType.i
# noinspection PyProtectedMember
def dynamo_load(self, values, *, context, **kwargs):
if values is None:
return []
load = DynamicType.i._load
return [
load(value, context=context, **kwargs)
for value in values]
def dynamo_dump(self, values, *, context, **kwargs):
if values is None:
return None
# noinspection PyProtectedMember
dump = guard_no_action(DynamicType.i._dump)
dumped = (dump(value, context=context, **kwargs) for value in values)
return [value for value in dumped if value is not None] or None
[docs]class DynamicMap(Type):
"""Holds a dictionary of arbitrary values, including other DynamicLists and DynamicMaps.
Similar to :class:`~bloop.types.Map` but is not constrained to a single type.
.. code-block:: python
value = {"f": 1, "in": [True]]
DynamicMap()._dump(value)
-> {"M": {"f": {"N": 1}, "in": {"L": [{"BOOL": true}]}}}
.. note::
Values will only be loaded and dumped as their DynamoDB backing types. This means datetimes and uuids are
stored and loaded as strings, and timestamps are stored and loaded as integers. For more information, see
:ref:`dynamic-documents`.
"""
python_type = collections.abc.Mapping
backing_type = MAP
def __getitem__(self, key):
"""Overload allows easy nested access to types"""
return DynamicType.i
def dynamo_load(self, values, *, context, **kwargs):
if values is None:
return {}
# noinspection PyProtectedMember
return {
key: DynamicType.i._load(value, context=context, **kwargs)
for (key, value) in values.items()
}
def dynamo_dump(self, values, *, context, **kwargs):
if values is None:
return None
dumped = {}
# noinspection PyProtectedMember
dump = guard_no_action(DynamicType.i._dump)
for key, value in values.items():
value = dump(value, context=context, **kwargs)
if value is not None:
dumped[key] = value
return dumped or None
DYNAMIC_TYPES = {
"S": String(),
"N": Number(),
"B": Binary(),
"BOOL": Boolean(),
"SS": Set(String),
"NS": Set(Number),
"BS": Set(Binary),
"M": DynamicMap(),
"L": DynamicList()
}