Esquema de estrella en SQLAlchemy

Question

Tengo una base de datos de arquitectura de esquema en estrella que quiero representar en SQLAlchemy. Ahora tengo el problema de cómo hacer esto de la mejor manera posible. En este momento tengo muchas propiedades con condiciones de unión personalizadas, porque los datos se almacenan en diferentes tablas. Sería bueno si fuera posible reutilizar las dimensiones para diferentes tablas de hechos, pero no he descubierto cómo se puede hacer eso bien.

Answer 1

Una tabla de hechos típica en un esquema en estrella contiene referencias de clave externa a todas las tablas de dimensiones, por lo que, por lo general, no habría necesidad de condiciones de unión personalizadas: se determinan automáticamente a partir de referencias de clave externa.

Por ejemplo, se vería un esquema en estrella con dos tablas de hechos:

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)

Pero supongamos que quiere reducir la repetitiva en cualquier caso. Crearía generadores locales para las clases de dimensión que se configuran en una tabla de hechos:

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)

en cuyo caso el uso sería como:

class FactOne(Base):
    ...

Store.add_dimension(FactOne)

Pero, hay un problema con eso. Suponiendo que las columnas de dimensión que está agregando son columnas de clave principal, la configuración del mapeador fallará ya que una clase necesita tener sus claves primarias configuradas antes de configurar la asignación. Asumiendo que estamos usando declarative (que verá a continuación tiene un buen efecto), para que este enfoque funcione, tendríamos que usar la función instrument_declarative () en lugar de la metaclase estándar:

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

Entonces, haríamos algo como:

class Store(object):
    # ...

class FactOne(object):
    __tablename__ = 'sales_fact_one'

Store.add_dimension(FactOne)

register_cls(Store, FactOne)

Si realmente tiene una buena razón para las condiciones de unión personalizada, siempre que haya algún patrón sobre cómo se crean esas condiciones, puede generar eso con su 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)

Pero lo mejor si estás en 2.6 es convertir add_dimension en un decorador de clase. Aquí hay un ejemplo con todo limpio:

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()