Quickstart: Patterns and Best Practices¶
Meta Configuration Pattern¶
from sqlathanor import declarative_base, Column, relationship, AttributeConfiguration
from sqlalchemy import Integer, String
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.associationproxy import association_proxy
BaseModel = declarative_base()
class User(BaseModel):
__tablename__ = 'users'
__serialization__ = [AttributeConfiguration(name = 'id',
supports_csv = True,
csv_sequence = 1,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None),
AttributeConfiguration(name = 'addresses',
supports_json = True,
supports_yaml = (True, True),
supports_dict = (True, False),
on_serialize = None,
on_deserialize = None),
AttributeConfiguration(name = 'hybrid',
supports_csv = True,
csv_sequence = 2,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None)]
AttributeConfiguration(name = 'keywords',
supports_csv = False,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None)]
AttributeConfiguration(name = 'python_property',
supports_csv = (False, True),
csv_sequence = 3,
supports_json = (False, True),
supports_yaml = (False, True),
supports_dict = (False, True),
on_serialize = None,
on_deserialize = None)]
id = Column('id',
Integer,
primary_key = True)
addresses = relationship('Address',
backref = 'user')
_hybrid = 1
@hybrid_property
def hybrid(self):
return self._hybrid
@hybrid.setter
def hybrid(self, value):
self._hybrid = value
@hybrid.expression
def hybrid(cls):
return False
keywords = association_proxy('keywords', 'keyword')
@property
def python_property(self):
return self._hybrid * 2
Declarative Configuration Pattern¶
from sqlathanor import declarative_base, Column, relationship
from sqlalchemy import Integer, String
BaseModel = declarative_base()
class User(BaseModel):
__tablename__ = 'users'
id = Column('id',
Integer,
primary_key = True,
supports_csv = True,
csv_sequence = 1,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None)
addresses = relationship('Address',
backref = 'user',
supports_json = True,
supports_yaml = (True, True),
supports_dict = (True, False),
on_serialize = None,
on_deserialize = None)
Serializing Model Instances¶
# For a SQLAlchemy Model Class named "User" with an instance named "user":
as_csv = user.to_csv() # CSV
as_json = user.to_json() # JSON
as_yaml = user.to_yaml() # YAML
as_dict = user.to_dict() # dict
Updating a Model Instance¶
See also
# For a SQLAlchemy Model Class named "User" with an instance named "user"
# and serialized objects "as_csv" (string), "as_json" (string),
# "as_yaml" (string), and "as_dict" (dict):
user.update_from_csv(as_csv) # CSV
user.update_from_json(as_json) # JSON
user.update_from_yaml(as_yaml) # YAML
user.update_from_dict(as_dict) # dict
Creating a New Model Instance¶
See also
# For a SQLAlchemy Model Class named "User" and serialized objects "as_csv"
# (string), "as_json" (string), "as_yaml" (string), and "as_dict" (dict):
user = User.new_from_csv(as_csv) # CSV
user = User.new_from_json(as_json) # JSON
user = User.new_from_yaml(as_yaml) # YAML
user = User.new_from_dict(as_dict) # dict
Error Handling¶
Errors During Serialization¶
from sqlathanor.errors import SerializableAttributeError, \
UnsupportedSerializationError, MaximumNestingExceededError
# For a SQLAlchemy Model Class named "User" and a model instance named "user".
try:
as_csv = user.to_csv()
as_json = user.to_json()
as_yaml = user.to_yaml()
as_dict = user.to_dict()
except SerializableAttributeError as error:
# Handle the situation where "User" model class does not have any attributes
# serializable to JSON.
pass
except UnsupportedSerializationError as error:
# Handle the situation where one of the "User" model attributes is of a data
# type that does not support serialization.
pass
except MaximumNestingExceededError as error:
# Handle a situation where "user.to_json()" received max_nesting less than
# current_nesting.
#
# This situation is typically an error on the programmer's part, since
# SQLAthanor by default avoids this kind of situation.
#
# Best practice is simply to let this exception bubble up.
raise error
Errors During De-serialization¶
from sqlathanor.errors import DeserializableAttributeError, \
CSVStructureError, DeserializationError, ValueDeserializationError, \
ExtraKeysError, UnsupportedDeserializationError
# For a SQLAlchemy Model Class named "User" and a model instance named "user",
# with serialized data in "as_csv", "as_json", "as_yaml", and "as_dict" respectively.
try:
user.update_from_csv(as_csv)
user.update_from_json(as_json)
user.update_from_yaml(as_yaml)
user.update_from_dict(as_dict)
new_user = User.new_from_csv(as_csv)
new_user = User.new_from_json(as_json)
new_user = User.new_from_yaml(as_yaml)
new_user = User.new_from_dict(as_dict)
except DeserializableAttributeError as error:
# Handle the situation where "User" model class does not have any attributes
# de-serializable from the given format (CSV, JSON, YAML, or dict).
pass
except DeserializationError as error:
# Handle the situation where the serialized object ("as_csv", "as_json",
# "as_yaml", "as_dict") cannot be parsed, for example because it is not
# valid JSON, YAML, or dict.
pass
except CSVStructureError as error:
# Handle the situation where the structure of "as_csv" does not match the
# expectation configured for the "User" model class.
raise error
except ExtraKeysError as error:
# Handle the situation where the serialized object ("as_json",
# "as_yaml", "as_dict") may have unexpected keys/attributes and
# the error_on_extra_keys argument is False.
#
# Applies to: *_from_json(), *_from_yaml(), and *_from_dict() methods
pass
except ValueDeserializationError as error:
# Handle the situation where an input value in the serialized object
# raises an exception in the deserialization post-processing function.
pass
except UnsupportedDeserializationError as error:
# Handle the situation where the de-serialization process attempts to
# assign a value to an attribute that does not support de-serialization.
pass
Password De-serialization¶
See also
from sqlathanor import declarative_base, Column, AttributeConfiguration
from sqlalchemy import Integer, String
def my_encryption_function(value):
"""Function that accepts an inbound password ``value`` and returns its
encrypted value."""
# ENCRYPTION LOGIC GOES HERE
return encrypted_value
BaseModel = declarative_base()
class User(BaseModel):
__tablename__ = 'users'
__serialization__ = [AttributeConfiguration(name = 'id',
supports_csv = True,
csv_sequence = 1,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None),
AttributeConfiguration(name = 'password',
supports_csv = (True, False),
supports_json = (True, False),
supports_yaml = (True, False),
supports_dict = (True, False),
on_serialize = None,
on_deserialize = my_encryption_function)]
id = Column('id',
Integer,
primary_key = True)
password = Column('password', String(255))
from sqlathanor import declarative_base, Column
from sqlalchemy import Integer, String
def my_encryption_function(value):
"""Function that accepts an inbound password ``value`` and returns its
encrypted value."""
# ENCRYPTION LOGIC GOES HERE
return encrypted_value
BaseModel = declarative_base()
class User(BaseModel):
__tablename__ = 'users'
id = Column('id',
Integer,
primary_key = True,
supports_csv = True,
csv_sequence = 1,
supports_json = True,
supports_yaml = True,
supports_dict = True,
on_serialize = None,
on_deserialize = None)
password = Column('password',
String(255),
supports_csv = (True, False),
csv_sequence = 2,
supports_json = (True, False),
supports_yaml = (True, False),
supports_dict = (True, False),
on_serialize = None,
on_deserialize = my_encryption_function)
Using SQLAthanor with Flask-SQLAlchemy¶
See also
- Import SQLAthanor > Using Flask-SQLAlchemy
- Flask-SQLAlchemy Documentation
from sqlathanor import FlaskBaseModel, initialize_sqlathanor
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(model_class = FlaskBaseModel)
db = initialize_sqlathanor(db)