# -*- coding: utf-8 -*-
# flake8: noqa
from __future__ import unicode_literals
import os
import warnings
from six.moves import builtins as __builtin__
import logging
import uuid
import inspect
import dateutil.parser
import datetime
from uuid import uuid4
from functools import partial
from decimal import Decimal
from collections import OrderedDict
import sqlalchemy as db
from sqlalchemy import (
create_engine,
MetaData,
)
from sqlalchemy import Numeric
MODEL_REGISTRY = OrderedDict()
MODELS_BY_TABLE = OrderedDict()
format_decimal = lambda num: '{0:.2f}'.format(num)
logger = logging.getLogger(__name__)
[docs]class Monetary(Numeric):
def __init__(self):
super().__init__(precision=10, scale=2)
def generate_uuid():
return bytes(uuid4().hex)
def now():
return datetime.datetime.utcnow()
def DefaultForeignKey(field_name, parent_field_name,
ondelete='CASCADE', nullable=False, **kw):
return db.Column(field_name, db.Integer,
db.ForeignKey(parent_field_name, ondelete=ondelete),
nullable=nullable, **kw)
def PrimaryKey(name='id'):
return db.Column(name, db.Integer, primary_key=True)
def AutoUUID(name='uuid'):
return db.Column(name, db.String(32), default=generate_uuid)
def is_builtin_model(target):
return target.__module__.startswith('chemist.') and target.__name__ in ('ORM', 'Model')
[docs]class ORM(type):
"""metaclass for :py:class:`chemist.models.Model`
"""
def __init__(cls, name, bases, attrs):
if is_builtin_model(cls):
return
if not is_builtin_model(cls) and not hasattr(cls, 'table'):
raise TypeError('{} must have a table attribute defined at the class level'.format(cls))
columns = {c.name: c.type.python_type
for c in cls.table.columns}
cls.__columns__ = columns
attrs['__columns__'] = columns
ORM.register_model_class(cls, columns)
super(ORM, cls).__init__(name, bases, attrs)
@staticmethod
def determine_model_identity(cls):
return '.'.join([cls.__module__, cls.__name__])
@staticmethod
def register_model_class(cls, columns):
class_id = ORM.determine_model_identity(cls)
MODEL_REGISTRY[class_id] = columns
MODELS_BY_TABLE[cls.table.name] = cls
return cls
@staticmethod
def get_columns_for_model_class(cls):
class_id = ORM.determine_model_identity(cls)
return MODEL_REGISTRY.get(class_id, {}) or {}
@staticmethod
def get_columns_for_model_instance(instance):
return ORM.get_columns_for_model_class(instance.__class__)
@staticmethod
def get_model_class_for_table_name(name):
return MODELS_BY_TABLE.get(name)
[docs]class Context(object):
"""Context is a system component that keeps track of multiple engines,
switch between them and manage their lifecycle.
It also provides a :py:class:`sqlalchemy.MetaData` instance that is automatically bound to
Its purpose is to leverage quicky swapping the engine between "unit" tests.
"""
def __init__(self, default_uri=None):
self.default_uri = default_uri or os.getenv('CHEMIST_SQLALCHEMY_URI')
self.engines = OrderedDict()
self.metadata = MetaData()
def set_default_uri(self, uri):
self.default_uri = uri
self.metadata.bind = self.get_default_engine()
return self.metadata
def get_or_create_engine(self, uri, *args, **kwargs):
engine = self.engines.get(uri, self.engines.get(uri)) or create_engine(uri, **kwargs)
self.engines[uri] = engine
return engine
@property
def engine(self):
return self.get_default_engine()
def get_engine(self, uri=None):
if self.uri:
return self.get_or_create_engine(uri)
return self.get_default_engine()
def get_default_engine(self):
return self.get_or_create_engine(self.default_uri)
def DefaultTable(self, name, *fields):
options = dict(
# mysql_engine='InnoDB',
# mysql_charset='utf8',
# mysql_key_block_size="1024",
)
table = db.Table(
name,
self.metadata,
*fields,
**options
)
return table
default_context = Context()
metadata = default_context.metadata
get_or_create_engine = default_context.get_or_create_engine
DefaultTable = default_context.DefaultTable
def get_first_available_engine():
for engine in filter(bool, default_context.engines.values()):
return engine
def get_engine(uri=None, key=None):
if not uri:
return default_context.get_default_engine()
if key:
warnings.warn("{} will be deprecated in the next minor version of chemist. Pass a key for a Context instead".format(key), DeprecationWarning)
return get_or_create_engine(uri=uri)
def set_default_uri(uri):
default_context.set_default_uri(uri)
return default_context.get_default_engine()