Skip to content

Instantly share code, notes, and snippets.

@miohtama
Created April 27, 2015 18:06
Show Gist options
  • Save miohtama/87594badd782f6233f99 to your computer and use it in GitHub Desktop.
Save miohtama/87594badd782f6233f99 to your computer and use it in GitHub Desktop.
JSOB field decoratore: Create SQLAlchemy model properties which are stored inside JSONB data
"""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)
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")
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")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment