Quickstart: Patterns and Best Practices


Installation

To install SQLAthanor, just execute:

$ pip install sqlathanor

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

# 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

# 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

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 SQLAlchemy Reflection

from sqlathanor import declarative_base, Column, AttributeConfiguration

from sqlalchemy import create_engine, Table

BaseModel = declarative_base()

engine = create_engine('... ENGINE CONFIGURATION GOES HERE ...')
# NOTE: Because reflection relies on a specific SQLAlchemy Engine existing, presumably
# you would know how to configure / instantiate your database engine using SQLAlchemy.
# This is just here for the sake of completeness.

class ReflectedUser(BaseModel):
    __table__ = Table('users',
                      BaseModel.metadata,
                      autoload = True,
                      autoload_with = engine)

    __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 = None)]

    # ADDITIONAL RELATIONSHIPS, HYBRID PROPERTIES, OR ASSOCIATION PROXIES
    # GO HERE
from sqlathanor import declarative_base, Column, AttributeConfiguration

from sqlalchemy import create_engine, Table, Integer, String

BaseModel = declarative_base()

engine = create_engine('... ENGINE CONFIGURATION GOES HERE ...')
# NOTE: Because reflection relies on a specific SQLAlchemy Engine existing, presumably
# you would know how to configure / instantiate your database engine using SQLAlchemy.
# This is just here for the sake of completeness.

UserTable = Table('users',
                  BaseModel.metadata,
                  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),
                  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 = None))

class ReflectedUser(BaseModel):
    __table__ = Table('users',
                      BaseModel.metadata,
                      autoload = True,
                      autoload_with = engine)

    # ADDITIONAL RELATIONSHIPS, HYBRID PROPERTIES, OR ASSOCIATION PROXIES
    # GO HERE

Tip

In practice, this pattern eliminates the time-saving benefits of using reflection in the first place. Instead, I would recommend adopting the meta configuration pattern with reflection instead.

from sqlathanor import declarative_base, Column, AttributeConfiguration

from sqlalchemy import create_engine, Table, Integer, String

BaseModel = declarative_base()

engine = create_engine('... ENGINE CONFIGURATION GOES HERE ...')
# NOTE: Because reflection relies on a specific SQLAlchemy Engine existing, presumably
# you would know how to configure / instantiate your database engine using SQLAlchemy.
# This is just here for the sake of completeness.

class ReflectedUser(BaseModel):
    __table__ = Table('users',
                      BaseModel.metadata,
                      autoload = True,
                      autoload_with = engine)

    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 = None)

    # ADDITIONAL RELATIONSHIPS, HYBRID PROPERTIES, OR ASSOCIATION PROXIES
    # GO HERE

Using SQLAthanor with Automap

New in version 0.2.0.

Error

If you try to use automap_base() with SQLAlchemy v.0.9.0, you will get a SQLAlchemySupportError.

from sqlathanor.automap import automap_base
from sqlalchemy import create_engine

# Create your Automap Base
Base = automap_base()

engine = create_engine('... DATABASE CONNECTION GOES HERE ...')

# Prepare your automap base. This reads your database and creates your models.
Base.prepare(engine, reflect = True)

# And here you can create a "User" model class and an "Address" model class.
User = Base.classes.users
Address = Base.classes.addresses


User.set_attribute_serialization_config('email_address',
                                        supports_csv = True,
                                        supports_json = True,
                                        supports_yaml = True,
                                        supports_dict = True)
User.set_attribute_serialization_config('password',
                                        supports_csv = (True, False),
                                        supports_json = (True, False),
                                        supports_yaml = (True, False),
                                        supports_dict = (True, False),
                                        on_deserialize = my_encryption_function)
from sqlathanor.automap import automap_base
from sqlalchemy import create_engine

# Create your Automap Base
Base = automap_base()

engine = create_engine('... DATABASE CONNECTION GOES HERE ...')

# Prepare your automap base. This reads your database and creates your models.
Base.prepare(engine, reflect = True)

# And here you can create a "User" model class and an "Address" model class.
User = Base.classes.users
Address = Base.classes.addresses

User.__serialization__ = [
    {
        'name': 'email_address',
        'supports_csv': True,
        'supports_json': True,
        'supports_yaml': True,
        'supports_dict': True
    },
    {
        'name': 'password',
        'supports_csv': (True, False),
        'supports_json': (True, False),
        'supports_yaml': (True, False),
        'supports_dict': (True, False),
        'on_deserialize': my_encryption_function
    }
]

Using SQLAthanor with Flask-SQLAlchemy

See also

from sqlathanor import FlaskBaseModel, initialize_sqlathanor
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy(model_class = FlaskBaseModel)
db = initialize_sqlathanor(db)