On May 11, 2014, at 12:27 AM, Joseph Casale <[email protected]> wrote:
> > > What I wanted to know was if it was possible to construct either a table > definition > for TableB so that someone could simply pass in actual values of table_a.name > to meta columns in table_b. For example if #3 above is not possible and > table_a > has been pre populated: > > table_a: > id name > -- ---- > 1 foo > 2 bar > 3 biz > > To populate table_b: > > data = [ > TableB(name='foo'), > TableB(name='bar') > TableB(name='biz') > ] > session.add_all(data) > > Of course table_b has ~13 columns for which many combinations of values from > all the intermediate tables will produce unique rows... I don't know about the "~13" columns part here but what you have is a combination of: 1. "unique object" recipe, https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/UniqueObject, which will give you TableA(somename) doing a "create if not exists" pattern, the ORM doesn't support all the various MERGE/INSERT OR REPLACE/OR IGNORE games that mostly MySQL plays so it's just a straight up SELECT 2. "association proxy", http://docs.sqlalchemy.org/en/latest/orm/extensions/associationproxy.html, so that you can link the mutation/getting of a plain scalar (string, int, etc.) attribute to that of a mapped object. def _unique(session, cls, hashfunc, queryfunc, constructor, arg, kw): cache = getattr(session, '_unique_cache', None) if cache is None: session._unique_cache = cache = {} key = (cls, hashfunc(*arg, **kw)) if key in cache: return cache[key] else: with session.no_autoflush: q = session.query(cls) q = queryfunc(q, *arg, **kw) obj = q.first() if not obj: obj = constructor(*arg, **kw) session.add(obj) cache[key] = obj return obj class UniqueMixin(object): @classmethod def unique_hash(cls, *arg, **kw): raise NotImplementedError() @classmethod def unique_filter(cls, query, *arg, **kw): raise NotImplementedError() @classmethod def as_unique(cls, *arg, **kw): session = Session() return _unique( session, cls, cls.unique_hash, cls.unique_filter, cls, arg, kw) from sqlalchemy import Column, Integer, ForeignKey, String, create_engine from sqlalchemy.orm import relationship, scoped_session, sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.associationproxy import association_proxy Base = declarative_base() class TableA(UniqueMixin, Base): __tablename__ = 'tablea' id = Column(Integer, primary_key=True) name = Column(String, unique=True) def __init__(self, name): self.name = name @classmethod def unique_hash(cls, name): return hash(name) @classmethod def unique_filter(cls, query, name): return query.filter(TableA.name == name) class TableB(Base): __tablename__ = 'tableb' id = Column(Integer, primary_key=True) name_id = Column(ForeignKey('tablea.id')) _name = relationship("TableA") name = association_proxy("_name", "name", creator=lambda name: TableA.as_unique(name)) engine = create_engine("sqlite://", echo=True) Base.metadata.create_all(engine) Session = scoped_session(sessionmaker(engine)) Session.add_all([ TableB(name='foo'), TableB(name='bar'), TableB(name='bat'), TableB(name='hoho'), TableB(name='bar'), TableB(name='hoho'), TableB(name='foo'), ]) Session.commit() -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To post to this group, send email to [email protected]. Visit this group at http://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.
