Quickstart: Patterns and Best Practices¶
- Installation
- Meta Configuration Pattern
- Declarative Configuration Pattern
- Serializing Model Instances
- Updating a Model Instance
- Creating a New Model Instance
- Error Handling
- Password De-serialization
- Programmatically Generating Models
- Using SQLAthanor with SQLAlchemy Reflection
- Using SQLAthanor with Automap
- Using SQLAthanor with Flask-SQLAlchemy
- Generating SQLAlchemy Tables Programmatically
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)
Programmatically Generating Models¶
New in version 0.3.0: generation from CSV, JSON, YAML, or dict
New in version 0.8.0: generation from Pydantic models
See also
# FROM CSV:
from sqlathanor import generate_model_from_csv
# Assuming that "csv_data" contains your CSV data
CSVModel = generate_model_from_csv(csv_data,
tablename = 'my_table_name',
primary_key = 'id')
from sqlathanor import generate_model_from_json
# Assuming that "json_string" contains your JSON data in a string
JSONModel = generate_model_from_json(json_string,
tablename = 'my_table_name',
primary_key = 'id')
from sqlathanor import generate_model_from_yaml
# Assuming that "yaml_string" contains your YAML data in a string
YAMLModel = generate_model_from_yaml(yaml_string,
tablename = 'my_table_name',
primary_key = 'id')
from sqlathanor import generate_model_from_dict
# Assuming that "yaml_string" contains your YAML data in a string
DictModel = generate_model_from_dict(dict_string,
tablename = 'my_table_name',
primary_key = 'id')
from sqlathanor import generate_model_from_pydantic
# Assumes that "PydanticReadModel" and "PydanticWriteModel" contain the Pydantic
# models for the object/resource you are representing.
PydanticModel = generate_model_from_pydantic([PydanticReadModel, PydanticWriteModel],
tablename = 'my_table_name',
primary_key = 'id')
Using SQLAthanor with SQLAlchemy Reflection¶
See also
- Using Declarative Reflection with SQLAthanor
- SQLAlchemy: Reflecting Database Objects
- SQLAlchemy: Using Reflection with Declarative
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.
See also
- Using Automap with SQLAthanor
- SQLAlchemy: Automap Extension
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
- Import SQLAthanor > Using Flask-SQLAlchemy
- Flask-SQLAlchemy Documentation
from sqlathanor import FlaskBaseModel, initialize_flask_sqlathanor
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app, model_class = FlaskBaseModel)
db = initialize_flask_sqlathanor(db)
Generating SQLAlchemy Tables Programmatically¶
New in version 0.3.0: CSV, JSON, YAML, and dict
support
New in version 0.8.0: Pydantic model support
See also
from sqlathanor import Table
# Assumes CSV data is in "csv_data" and a MetaData object is in "metadata"
csv_table = Table.from_csv(csv_data,
'tablename_goes_here',
metadata,
'primary_key_column',
column_kwargs = None,
skip_nested = True,
default_to_str = False,
type_mapping = None)
from sqlathanor import Table
# Assumes JSON string is in "json_data" and a MetaData object is in "metadata"
json_table = Table.from_json(json_data,
'tablename_goes_here',
metadata,
'primary_key_column',
column_kwargs = None,
skip_nested = True,
default_to_str = False,
type_mapping = None)
from sqlathanor import Table
# Assumes YAML string is in "yaml_data" and a MetaData object is in "metadata"
yaml_table = Table.from_yaml(yaml_data,
'tablename_goes_here',
metadata,
'primary_key_column',
column_kwargs = None,
skip_nested = True,
default_to_str = False,
type_mapping = None)
from sqlathanor import Table
# Assumes dict object is in "dict_data" and a MetaData object is in "metadata"
dict_table = Table.from_dict(dict_data,
'tablename_goes_here',
metadata,
'primary_key_column',
column_kwargs = None,
skip_nested = True,
default_to_str = False,
type_mapping = None)
New in version 0.8.0.
from pydantic import BaseModel
from sqlathanor import Table
# Define Your Pydantic Models
class UserWriteModel(BaseModel):
id: int
username: str
email: str
password: str
class UserReadModel(BaseModel):
id: int
username: str
email: str
# Create Your Table
pydantic_table = Table.from_pydantic([UserWriteModel, UserReadModel],
tablename = 'my_tablename_goes_here',
primary_key = 'id')