schema in one place, abstract sqlalchemy vs django

coincharts/daemon/base.py

@@ -19,7 +19,7 @@
import daemon
import daemon.pidfile

from mutils import simple_alchemy
from coincharts import schema

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@@ -45,22 +45,6 @@ class PriceSeries(object):
    date_format_template = '%Y-%m-%dT%H:%M:%S.%f0Z'  # magic
    headers = {'X-CoinAPI-Key':
               open(API_KEY_FILE).read().strip()}
    schema = [
        ('time_period_start', 'TEXT'),
        ('time_period_end', 'TEXT'),
        ('time_open', 'TEXT'),
        ('time_close', 'TEXT'),
        ('price_open', 'REAL'),
        ('price_high', 'REAL'),
        ('price_low', 'REAL'),
        ('price_close', 'REAL'),
        ('volume_traded', 'REAL'),
        ('trades_count', 'INTEGER'),
    ]

    # these are THE values we care about, so CAPS
    TIME = 'time_period_end'
    PRICE = 'price_close'

    # this is the beginning of time if we don't have any local data
    first_date = '2018-01-09T00:00:00.0000000Z'
@@ -73,7 +57,7 @@ def get_db_session(cls, dir_path):
        if cls._session is not None:
            return cls._session
        db_path = os.path.join(os.path.abspath(dir_path), 'db.sqlite3')
        cls._session = simple_alchemy.get_session(db_path)
        cls._session = schema.get_sqlalchemy_session(db_path)
        return cls._session

    @classmethod
@@ -101,19 +85,7 @@ def __init__(self, symbol_id, dir_path):
        logger.debug('creating PriceSeries object for {}'.format(symbol_id))
        self.symbol_id = symbol_id
        self.dir_path = dir_path
        schema = [
            ('time_period_start', String),
            ('time_period_end', String),
            ('time_open', String),
            ('time_close', String),
            ('price_open', Float),
            ('price_high', Float),
            ('price_low', Float),
            ('price_close', Float),
            ('volume_traded', Float),
            ('trades_count', Integer),
        ]
        self.data = simple_alchemy.get_table_class(symbol_id, schema=schema)
        self.data = schema.get_db_table(symbol_id, 'sqlalchemy')

    def get_prices_since(self, start_dt):
        """Get prices for this PriceSeries where datetime is greater or equal to `start_dt`
@@ -125,7 +97,7 @@ def get_prices_since(self, start_dt):
        except TypeError:
            pass
        start_dt = self.get_normalized_datetime(self.round_up_hour(start_dt))
        kwargs = {self.TIME: start_dt}
        kwargs = {schema.datetime_field: start_dt}
        session = self.get_db_session(self.dir_path)
        first = session.query(self.data).filter_by(**kwargs).first()
        results = session.query(self.data).filter(id >= first).first()
@@ -180,7 +152,7 @@ def get_last_date_from_store(self):
        obj = session.query(self.data).order_by(self.data.id.desc()).first()
        if obj is None:
            return parse_dt(self.first_date)
        dt = getattr(obj, self.TIME)
        dt = getattr(obj, schema.datetime_field)
        return parse_dt(dt)

    def insert(self, data):

coincharts/schema.py

@@ -0,0 +1,129 @@
# -*- mode: python; coding: utf-8 -*-
"""Abstraction for sqlalchemy vs django engines.
"""

# It may be better/easier to just use django in place of sqlalchemy. TBD.

import os
from types import MethodType
import atexit

import django.db.models
import sqlalchemy.types
import sqlalchemy.ext.declarative

datetime_field = 'time_period_end'
price_field = 'price_close'

schema = [
    ('time_period_start', 'TEXT'),
    ('time_period_end', 'TEXT'),
    ('time_open', 'TEXT'),
    ('time_close', 'TEXT'),
    ('price_open', 'REAL'),
    ('price_high', 'REAL'),
    ('price_low', 'REAL'),
    ('price_close', 'REAL'),
    ('volume_traded', 'REAL'),
    ('trades_count', 'INTEGER'),
]

type_mappings = {
    'django': {
        'TEXT': django.db.models.CharField,
        'REAL': django.db.models.FloatField,
        'INTEGER': django.db.models.IntegerField,
    },
    
    'sqlalchemy': {
        'TEXT': sqlalchemy.types.String,
        'INTEGER': sqlalchemy.types.Integer,
        'REAL': sqlalchemy.types.Float,
    },
}

# to-be-bound to classes created below
def __str__(self):
    class_name = self.__class__.__name__
    return '<{}({}@{})>'.format(
        class_name,
        getattr(self, TIME),
        getattr(self, PRICE),
    )
Base = None

def get_db_table(symbol_id, orm):
    global Base
    mapping = type_mappings[orm]
    columns = {}
    for name, type_ in schema:
        type_ = mapping[type_]
        if orm == 'django':
            columns[name] = type_()
        elif orm == 'sqlalchemy':
            columns[name] = sqlalchemy.Column(type_)
        else:
            raise Exception('Unknown ORM "{}"'.format(orm))

    if orm == 'django':
        klass = type(symbol_id, (models.Model,), columns)
        klass.__str__ = __str__
    elif orm == 'sqlalchemy':
        Base = sqlalchemy.ext.declarative.declarative_base()
        columns['__tablename__'] = symbol_id.lower()
        columns['id'] = sqlalchemy.Column(sqlalchemy.types.Integer, primary_key=True)
        klass = type(symbol_id, (Base,), columns)
        klass.__repr__ = __str__
    else:
        raise Exception('fixme: tooooo complicated')
    
    return klass


def get_sqlalchemy_session(db_path, echo=False):
    """Create and return a SQLAlchemy `session` object
    """
    db_path = os.path.abspath(db_path)
    engine = sqlalchemy.create_engine('sqlite:///{db_path}'.format(db_path=db_path), echo=echo)
    Session = sqlalchemy.orm.sessionmaker(bind=engine)
    session = Session()
    Base.metadata.create_all(engine)
    atexit.register(session.close)
    return session


"""
class Page(models.Model):
    weight = models.FloatField(default=0.0)
    name = models.CharField(max_length=30)
    label = models.CharField(max_length=60)
    headline = models.CharField(max_length=200)
    tab_visible = models.BooleanField('Should tab be visible?', default=True)

    def __str__(self):
        return self.name

class Content(models.Model):
    page = models.ForeignKey(Page, on_delete=models.CASCADE)
    main = models.TextField(blank=True)
    rhs = models.TextField(blank=True)

    def __str__(self):
        return self.page.name
"""


"""
from sqlalchemy import Column, Integer, String
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
       return "<User(name='%s', fullname='%s', password='%s')>" % (
                            self.name, self.fullname, self.password)
"""