JSOB field decoratore: Create SQLAlchemy model properties which are stored inside JSONB data

jsonb.py

"""JSONB data utilities."""
import datetime
from decimal import Decimal
import copy
import json

import iso8601

from sqlalchemy import inspect
from sqlalchemy.orm.attributes import flag_modified
from sqlalchemy.orm.attributes import set_attribute


import jsonpointer


class BadJSONData(Exception):
    """You tried to save something which cannot be stick into JSON data."""


class BadStoredData(Exception):
    """We found something which is non-JSON like from the database."""


class CannotLookupData(Exception):
    """Member get failed."""

class CannotProcessISO8601(Exception):
    """You passed in a naive datetime without explicit timezone information."""


class NullConverter(object):

    def serialize(self, val):
        return val

    def deserialize(self, val):
        return val


class ISO8601DatetimeConverter(object):
    """Serialize datetime to ISO8601 unless it's None."""

    def serialize(self, val):

        if val is None:
            return val

        assert isinstance(val, datetime.datetime), "Got type {}".format(type(val))

        if not val.tzinfo:
            raise CannotProcessISO8601("datetime lacks timezone information: {}. Please use datetime.datetime.utcnow(datetime.timezone.utc)".format(val))

        return val.isoformat()

    def deserialize(self, val):

        if val is None:
            return val

        return iso8601.parse_date(val)


_marker = object()


class JSONBProperty(object):
    """Define a Python class property which can set/get JSONB field data.

    The location of the propety is defined using JSON Pointer notation.

    - `JSON Pointer Notation specification <https://tools.ietf.org/html/rfc6901>`_

    - Underlying `jsonpointer Python module documentation <http://python-json-pointer.readthedocs.org/en/latest/index.html>`_

    In the case the field value is not yet set and you try to access it, a Falsy valud ``JSONBField.UNDEFINED` is returned.

    You must be aware of JSON data limitations when using JSONB properties

    - All numbers are floats

    - No native support for datetime storage

    :param data_field: The name of the SQLAlchemy field holding JSONB data on this model.

    :param pointer: RFC6901 pointer to the field.

    :param converter: JSON serializer/deserializer. Can be a class or instance with serialize() / deserialize() methods.

    :param graceful: If set, return this value when the member is not found instead of raising exception
    """

    #: Return this value if there is nothing at the end of RFC 6901 pointer
    UNDEFINED = jsonpointer._nothing

    def __init__(self, data_field, pointer, graceful=_marker, converter=NullConverter):
        self.data_field = data_field
        self.pointer = pointer

        # Passed in a class
        if type(converter) == type:
            converter = converter()

        self.converter = converter
        self.graceful = graceful

    def ensure_valid_data(self, obj):
        """Handle freshly created objects and corrupted data more gracefully."""

        field = getattr(obj, self.data_field)

        if field is None:
            insp = inspect(obj)

            if insp.persistent or insp.detached:
                raise BadStoredData("Field {} contained None (NULL) - make sure it is initialized with empty dictionary".format(self.data_field))
            else:
                # An object of which field default value is not yet set
                data = {}
                setattr(obj, self.data_field, data)
                return data

        return field

    def is_graceful(self):
        return self.graceful != _marker

    def __get__(self, obj, objtype=None):
        data = self.ensure_valid_data(obj)
        try:
            val = jsonpointer.resolve_pointer(data, self.pointer, default=JSONBProperty.UNDEFINED)
        except jsonpointer.JsonPointerException as e:
            # TODO: update jsonpointer to give more specific errors, this could vbe something else besides member not found
            if self.is_graceful():
                val = self.graceful
            else:
                raise CannotLookupData("Could not find {} on data {}".format(self.pointer, obj))
        return self.converter.deserialize(val)

    def __set__(self, obj, val):

        # TODO: Abstract JsonPointerException when settings a member with missing nested parent dict

        # We need to deepcopy data, so that SQLAlchemy can detect changes. Otherwise nested changes would mutate the dict in-place and SQLAlchemy cannot perform comparison.
        data = copy.deepcopy(self.ensure_valid_data(obj))

        val = self.converter.serialize(val)

        if val is not None:

            # Do some basic data validation, don't let first class objects slip through
            if type(val) not in (str, float, bool, int, dict):
                raise BadJSONData("Cannot update field at {} as it has unsupported type {} for JSONB data".format(self.pointer, type(val)))

        jsonpointer.set_pointer(data, self.pointer, val)

        set_attribute(obj, self.data_field, data)
        # flag_modified(obj, self.data_field)

other_examples.py

class Delivery(Base):

    __tablename__ = "delivery"

    id = Column(Integer, autoincrement=True, primary_key=True)

    uuid = Column(UUID(as_uuid=True))

    customer_id = Column(ForeignKey("users.id"))
    customer = relationship(User, primaryjoin=customer_id == User.id, backref="deliveries")

    #: Product id
    product = Column(String(32))

    delivery_status = Column(Enum('waiting', 'picked', 'delivered', 'cancelled', 'rescheduled', name="delivery_status"), nullable=False)

    #: When this account was created
    created_at = Column(DateTime(timezone=utc), default=now)

    #: When the account data was updated last time
    updated_at = Column(DateTime(timezone=utc), onupdate=now)

    estimated_delivery_at = Column(DateTime(timezone=utc), nullable=True)

    delivery_started_at = Column(DateTime(timezone=utc), nullable=True)

    delivery_signed_off_at = Column(DateTime(timezone=utc), nullable=True)

    delivery_cancelled_at = Column(DateTime(timezone=utc), nullable=True)

    #: Contains raw data from web form
    delivery_data = sqlalchemy.Column(JSONB, default={})

    street_address = JSONBProperty("delivery_data", "/delivery_info/street_address")

    postal_information = JSONBProperty("delivery_data", "/delivery_info/postal_information")

    phone_number = JSONBProperty("delivery_data", "/delivery_info/phone_number")

    email = JSONBProperty("delivery_data", "/delivery_info/email")

    full_name = JSONBProperty("delivery_data", "/delivery_info/full_name")

usermixin.py

from sqlalchemy import inspection
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import INET
from sqlalchemy import DateTime
from sqlalchemy import Column
from sqlalchemy import String
from sqlalchemy import Boolean
from sqlalchemy import Integer
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.mutable import MutableDict

import colander

from pyramid_web20.models import now
from pyramid_web20.models import utc
from pyramid_web20.models import DBSession
from pyramid_web20.utils.jsonb import JSONBProperty


#: Initialze user_data JSONB structure with these fields
DEFAULT_USER_DATA = {
    "full_name": None,
    "user_registration_source": None,

    # Is it the first time this user is logging to our system? If it is then take the user to fill in the profile page.
    "first_login": True,

    "social": {
    }
}


# TODO: "user" is reserved name on PSQL. File an issue against Horus.
# UserMixin.__tablename__ = "users"


class UserMixin:
    """A user who signs up with email or with email from social media.

    TODO: Make user user.id is not exposed anywhere e.g. in email activation links.
    """
    #: A test user
    USER_MEDIA_DUMMY = "dummy"

    #: Self sign up
    USER_MEDIA_EMAIL = "email"

    #: User signed up with Facebook
    USER_MEDIA_FACEBOOK = "facebook"

    #: User signed up with Github
    USER_MEDIA_GITHUB = "github"

    #: Admin group name
    GROUP_ADMIN = "admin"

    #: Though not displayed on the site, the concept of "username" is still preversed. If the site needs to have username (think Instragram, Twitter) the user is free to choose this username after the sign up. Username is null until the initial user activation is completed after db.flush() in create_activation().
    username = Column(String(32), nullable=True, unique=True)

    #: Store the salted password. This is nullable because users using social media logins may never set the password.
    _password = Column('password', String(256), nullable=True)

    #: The salt used for the password. Null if no password set.
    salt = Column(String(256), nullable=True)

    #: When this account was created
    created_at = Column(DateTime(timezone=utc), default=now)

    #: When the account data was updated last time
    updated_at = Column(DateTime(timezone=utc), onupdate=now)

    #: When this user was activated: email confirmed or first social login
    activated_at = Column(DateTime(timezone=utc), nullable=True)

    #: Is this user account enabled. The support can disable the user account in the case of suspected malicious activity.
    enabled = Column(Boolean, default=True)

    #: When this user accessed the system last time. None if the user has never logged in (only activation email sent). Information stored for the security audits.
    last_login_at = Column(DateTime(timezone=utc), nullable=True)

    #: From which IP address did this user log in from. If this IP is null the user has never logged in (only activation email sent). Information stored for the security audits. It is also useful for identifying the source country of users e.g. for localized versions.
    last_login_ip = Column(INET, nullable=True,
              info={'colanderalchemy': {
                        'typ': colander.String(),
                    }},
            )

    #: When this user changed the password for the last time. The value is null if the user comes from social networks. Information stored for the security audits.
    last_password_change_at = Column(DateTime, nullable=True)

    #: Store all user related settings in this expandable field
    user_data = Column((JSONB, default=DEFAULT_USER_DATA,
                    info={'colanderalchemy': {
                        'typ': colander.String(),
                    }},
            )

    #: Full name of the user (if given)
    full_name = JSONBProperty("user_data", "/full_name")

    #: How this user signed up to the site. May include string like "email", "facebook"
    registration_source = JSONBProperty("user_data", "/registration_source")

    #: Social media data of the user as a dict keyed by user media
    social = JSONBProperty("user_data", "/social")

    #: Is this the first login the user manages to do to our system. Use this information to redirect to special help landing page.
    first_login = JSONBProperty("user_data", "/first_login")