Звездная схема в SQLAlchemy

Question

У меня есть база данных с архитектурой звездной схемы, которую я хочу представить в SQLAlchemy. Теперь у меня проблема с тем, как это можно сделать наилучшим образом. Прямо сейчас у меня есть много свойств с пользовательскими условиями соединения, потому что данные хранятся в разных таблицах. Было бы неплохо, если бы можно было повторно использовать измерения для разных таблиц фактов, но я не понял, как это можно сделать красиво.

Answer 1

Типичная таблица фактов в звездообразной схеме содержит ссылки внешнего ключа на все таблицы измерений, поэтому обычно не требуется никаких пользовательских условий соединения - они определяются автоматически из ссылок внешнего ключа.

Например, схема типа «звезда» с двумя таблицами фактов будет выглядеть так:

Base = declarative_meta()

class Store(Base):
    __tablename__ = 'store'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(50), nullable=False)

class Product(Base):
    __tablename__ = 'product'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(50), nullable=False)

class FactOne(Base):
    __tablename__ = 'sales_fact_one'

    store_id = Column('store_id', Integer, ForeignKey('store.id'), primary_key=True)
    product_id = Column('product_id', Integer, ForeignKey('product.id'), primary_key=True)
    units_sold = Column('units_sold', Integer, nullable=False)

    store = relation(Store)
    product = relation(Product)

class FactTwo(Base):
    __tablename__ = 'sales_fact_two'

    store_id = Column('store_id', Integer, ForeignKey('store.id'), primary_key=True)
    product_id = Column('product_id', Integer, ForeignKey('product.id'), primary_key=True)
    units_sold = Column('units_sold', Integer, nullable=False)

    store = relation(Store)
    product = relation(Product)

Но предположим, что вы хотите уменьшить шаблон в любом случае. Я бы создал генераторы, локальные для классов измерений, которые настраивают себя на таблицу фактов:

class Store(Base):
    __tablename__ = 'store'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(50), nullable=False)

    @classmethod
    def add_dimension(cls, target):
        target.store_id = Column('store_id', Integer, ForeignKey('store.id'), primary_key=True)
        target.store = relation(cls)

В этом случае использование будет выглядеть следующим образом:

class FactOne(Base):
    ...

Store.add_dimension(FactOne)

Но есть проблема с этим. Предполагая, что добавляемые вами столбцы измерений являются столбцами первичного ключа, конфигурация преобразователя завершится сбоем, поскольку классу необходимо настроить свои первичные ключи перед настройкой сопоставления. Таким образом, предполагая, что мы используем декларативный (что вы увидите ниже, имеет хороший эффект), чтобы этот подход работал, нам пришлось бы использовать функцию instrument_declarative() вместо стандартного метакласса:

meta = MetaData()
registry = {}
def register_cls(*cls):
    for c in cls:
        instrument_declarative(c, registry, meta)

Итак, мы бы сделали что-то вроде:

class Store(object):
    # ...

class FactOne(object):
    __tablename__ = 'sales_fact_one'

Store.add_dimension(FactOne)

register_cls(Store, FactOne)

Если у вас действительно есть веская причина для пользовательских условий соединения, если есть какой-то шаблон для создания этих условий, вы можете сгенерировать его с помощью add_dimension():

class Store(object):
    ...

    @classmethod
    def add_dimension(cls, target):
        target.store_id = Column('store_id', Integer, ForeignKey('store.id'), primary_key=True)
        target.store = relation(cls, primaryjoin=target.store_id==cls.id)

Но последняя крутая вещь, если вы используете 2.6, это превратить add_dimension в декоратор классов. Вот пример со всем очищенным:

from sqlalchemy import *
from sqlalchemy.ext.declarative import instrument_declarative
from sqlalchemy.orm import *

class BaseMeta(type):
    classes = set()
    def __init__(cls, classname, bases, dict_):
        klass = type.__init__(cls, classname, bases, dict_)
        if 'metadata' not in dict_:
            BaseMeta.classes.add(cls)
        return klass

class Base(object):
    __metaclass__ = BaseMeta
    metadata = MetaData()
    def __init__(self, **kw):
        for k in kw:
            setattr(self, k, kw[k])

    @classmethod
    def configure(cls, *klasses):
        registry = {}
        for c in BaseMeta.classes:
            instrument_declarative(c, registry, cls.metadata)

class Store(Base):
    __tablename__ = 'store'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(50), nullable=False)

    @classmethod
    def dimension(cls, target):
        target.store_id = Column('store_id', Integer, ForeignKey('store.id'), primary_key=True)
        target.store = relation(cls)
        return target

class Product(Base):
    __tablename__ = 'product'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(50), nullable=False)

    @classmethod
    def dimension(cls, target):
        target.product_id = Column('product_id', Integer, ForeignKey('product.id'), primary_key=True)
        target.product = relation(cls)
        return target

@Store.dimension
@Product.dimension
class FactOne(Base):
    __tablename__ = 'sales_fact_one'

    units_sold = Column('units_sold', Integer, nullable=False)

@Store.dimension
@Product.dimension
class FactTwo(Base):
    __tablename__ = 'sales_fact_two'

    units_sold = Column('units_sold', Integer, nullable=False)

Base.configure()

if __name__ == '__main__':
    engine = create_engine('sqlite://', echo=True)
    Base.metadata.create_all(engine)

    sess = sessionmaker(engine)()

    sess.add(FactOne(store=Store(name='s1'), product=Product(name='p1'), units_sold=27))
    sess.commit()