def transaction_meta_factory(self):
        """
        Creates TransactionMeta class.
        """
        class TransactionMeta(
            self.declarative_base,
            TransactionMetaBase
        ):
            __tablename__ = 'transaction_meta'

        TransactionMeta.transaction_log = sa.orm.relationship(
            self.transaction_log_cls,
            backref=sa.orm.backref(
                'meta_relation',
                collection_class=attribute_mapped_collection('key')
            ),
            primaryjoin=(
                '%s.id == TransactionMeta.transaction_id' %
                self.transaction_log_cls.__name__
            ),
            foreign_keys=[TransactionMeta.transaction_id]
        )

        self.transaction_log_cls.meta = association_proxy(
            'meta_relation',
            'value',
            creator=lambda key, value: TransactionMeta(key=key, value=value)
        )

        return TransactionMeta

    def test_merge_irregular_collection(self):
        users, Address, addresses, User = (
            self.tables.users,
            self.classes.Address,
            self.tables.addresses,
            self.classes.User,
        )

        mapper(
            User,
            users,
            properties={
                "addresses": relationship(
                    mapper(Address, addresses),
                    backref="user",
                    collection_class=attribute_mapped_collection("email_address"),
                )
            },
        )
        u1 = User(id=7, name="fred")
        u1.addresses["[email protected]"] = Address(email_address="[email protected]")
        sess = create_session()
        sess.merge(u1)
        sess.flush()
        assert list(u1.addresses.keys()) == ["[email protected]"]

	def InitMapper( cls, metadata, Parameter, ParameterType, DesignQuantity ):
		mapper( cls, inherits = Parameter, polymorphic_identity = ParameterType, properties = {
			'_quantity' : relation( DesignQuantity,
				collection_class = attribute_mapped_collection('design'))
			})

		cls.quantity = association_proxy('_quantity', 'quantity', creator = lambda k, v: DesignQuantity( design = k, quantity = v ) )

	def InitMapper( cls, metadata, Parameter, ParameterType, ResourceQuantity ):
		mapper( cls, inherits = Parameter, polymorphic_identity = ParameterType, properties = {
			'_quantity' : relation( ResourceQuantity,
				collection_class = attribute_mapped_collection('resource'))
			})

		cls.quantity = association_proxy('_quantity', 'quantity', creator = lambda k, v: ResourceQuantity( resource = k, **v ) )

    def _variables(cls):
        # Camelcase the tablename to give the Variable inner class
        # here a specific class name; necessary for reporting on
        # classes
        class_name = \
            "".join(x.title() for x in cls.vars_tablename[:-1].split('_'))

        # Because we are constructing Variable inner class with the
        # 3-arg `type` function, we need to pull out all SA columns
        # given that initialization order matters for SA!
        #
        # * Defines the primary key with correct ordering
        # * Captures references, as seen in _repr_columns
        parent_id = Column(ForeignKey(
            '%s.id' % cls.__tablename__), primary_key=True)
        key = Column(String(255), primary_key=True)
        value = Column(JSONType)
        Variable = type(class_name, (Base,), {
            '__tablename__': cls.vars_tablename,
            'parent_id': parent_id,
            'key': key,
            'value': value,
            '_repr_columns': [key, value]})

        # Need a reference for the association proxy to lookup the
        # Variable class so it can reference
        cls.variable_class = Variable

        return relationship(
            Variable,
            collection_class=attribute_mapped_collection('key'),
            cascade='all, delete-orphan', lazy="joined")

 def test_merge_irregular_collection(self):
     mapper(User, users, properties={
         'addresses': relationship(
             mapper(Address, addresses),
             backref='user',
             collection_class=attribute_mapped_collection('email_address')),
         })
     u1 = User(id=7, name='fred')
     u1.addresses['[email protected]'] = Address(email_address='[email protected]')
     sess = create_session()
     sess.merge(u1)
     sess.flush()
     assert u1.addresses.keys() == ['[email protected]']

    def test_validator_bulk_dict_set(self):
        users, addresses, Address = (
            self.tables.users,
            self.tables.addresses,
            self.classes.Address,
        )

        class User(fixtures.ComparableEntity):
            @validates("addresses", include_removes=True)
            def validate_address(self, key, item, remove):
                if not remove:
                    assert isinstance(item, str)
                else:
                    assert isinstance(item, Address)
                item = Address(email_address=item)
                return item

        mapper(
            User,
            users,
            properties={
                "addresses": relationship(
                    Address,
                    collection_class=collections.attribute_mapped_collection(
                        "email_address"
                    ),
                )
            },
        )
        mapper(Address, addresses)

        u1 = User()
        u1.addresses["e1"] = "e1"
        u1.addresses["e2"] = "e2"
        eq_(
            u1.addresses,
            {
                "e1": Address(email_address="e1"),
                "e2": Address(email_address="e2"),
            },
        )
        u1.addresses = {"e3": "e3", "e4": "e4"}
        eq_(
            u1.addresses,
            {
                "e3": Address(email_address="e3"),
                "e4": Address(email_address="e4"),
            },
        )

def setup():
    """
    Set up ORM.

    Does not create any database tables, see :py:func:`create_tables`
    for that.
    """
    global resource_metadata_table
    if resource_metadata_table is None:
        log.debug('Defining resource metadata table')
        resource_metadata_table = Table(
            RESOURCE_METADATA_TABLE_NAME,
            metadata,
            Column('resource_id', types.UnicodeText, ForeignKey('resource.id',
                   ondelete='CASCADE', onupdate='CASCADE'), nullable=False,
                   primary_key=True),
            Column('last_extracted', types.DateTime),
            Column('last_url', types.UnicodeText),
            Column('last_format', types.UnicodeText),
            Column('task_id', types.UnicodeText)
        )
        mapper(
            ResourceMetadata,
            resource_metadata_table,
            properties={
                '_meta': relationship(ResourceMetadatum, collection_class=
                                      attribute_mapped_collection('key'),
                                      cascade='all, delete, delete-orphan'),
            }
        )
    else:
        log.debug('Resource metadata table already defined')
    global resource_metadatum_table
    if resource_metadatum_table is None:
        log.debug('Defining resource metadatum table')
        resource_metadatum_table = Table(
            RESOURCE_METADATUM_TABLE_NAME,
            metadata,
            Column('id', types.Integer, nullable=False, primary_key=True),
            Column('resource_id', types.UnicodeText, ForeignKey(
                   RESOURCE_METADATA_TABLE_NAME + '.resource_id',
                   ondelete='CASCADE', onupdate='CASCADE'), nullable=False),
            Column('key', types.UnicodeText, nullable=False),
            Column('value', types.UnicodeText)
        )
        mapper(ResourceMetadatum, resource_metadatum_table)
    else:
        log.debug('Resource metadatum table already defined')

    def test_attribute_mapped_collection(self):
        users, addresses = self.tables.users, self.tables.addresses

        mapper(User, users, properties={
            'addresses': relationship(
                Address,
                collection_class=attribute_mapped_collection('email_address')
            )
        })
        mapper(Address, addresses)
        u1 = User()
        u1.addresses = {"email1": Address(email_address="email1")}
        for loads, dumps in picklers():
            repickled = loads(dumps(u1))
            eq_(u1.addresses, repickled.addresses)
            eq_(repickled.addresses['email1'],
                Address(email_address="email1"))

    def variable_association(cls):
        name = cls.__name__
        discriminator = name.lower()

        # Defines a polymorphic class to distinguish variables stored
        # for regions, cells, etc.
        cls.variable_assoc_cls = assoc_cls = type(
            "%sVariableAssociation" % name,
            (VariableAssociation,),
            {
                '__tablename__': None,  # because mapping into a shared table
                '__mapper_args__': {
                    'polymorphic_identity': discriminator
                }
            })

        def _assoc_creator(kv):
            assoc = assoc_cls()
            for key, value in kv.items():
                assoc.variables[key] = Variable(key=key, value=value)
            return assoc

        cls._variables = association_proxy(
            'variable_association', 'variables', creator=_assoc_creator)

        # Using a composite associative proxy here enables returning the
        # underlying values for a given key, as opposed to the
        # Variable object; we need both.
        cls.variables = association_proxy(
            'variable_association', 'values', creator=_assoc_creator)

        def with_characteristic(self, key, value):
            return self._variables.any(key=key, value=value)

        cls.with_characteristic = classmethod(with_characteristic)

        rel = relationship(
            assoc_cls,
            collection_class=attribute_mapped_collection('key'),
            cascade='all, delete-orphan', lazy='joined',
            single_parent=True,
            backref=backref('parent', uselist=False))

        return rel

    def setup(self):
        class B(object):
            def __init__(self, key, elem):
                self.key = key
                self.elem = elem

        class A(object):
            elements = association_proxy("orig", "elem", creator=B)

        m = MetaData()
        a = Table('a', m, Column('id', Integer, primary_key=True))
        b = Table('b', m, Column('id', Integer, primary_key=True),
                    Column('aid', Integer, ForeignKey('a.id')))
        mapper(A, a, properties={
            'orig':relationship(B, collection_class=attribute_mapped_collection('key'))
        })
        mapper(B, b)
        self.A = A
        self.B = B

    def assign_translations(self):
        """
        Assigns translations relationship for translatable model. The assigned
        attribute is a relationship to all translation locales.
        """
        mapper = sa.orm.class_mapper(self.parent_cls)
        if not mapper.has_property('_translations'):
            foreign_keys = [
                getattr(self.translation_cls, column_key)
                for column_key in get_primary_keys(self.parent_cls).keys()
            ]

            mapper.add_property('_translations', sa.orm.relationship(
                self.translation_cls,
                primaryjoin=sa.and_(*self.primary_key_conditions),
                foreign_keys=foreign_keys,
                collection_class=attribute_mapped_collection('locale'),
                comparator_factory=TranslationComparator,
                cascade='all, delete-orphan',
                passive_deletes=option(self.parent_cls, 'passive_deletes'),
            ))

	def InitMapper( cls, metadata, Design, Component ):
		cls.__table__ = Table( cls.__tablename__, metadata,
				Column('design_id',    ForeignKey( Design.id ), primary_key = True ),
				Column('component_id', ForeignKey( Component.id ), primary_key = True ),
				Column('amount',       Integer, nullable = False, default = 1 ))

		cols = cls.__table__.c

		Index('ix_%s_design_component' % cls.__tablename__, cols.design_id, cols.component_id)

		mapper( cls, cls.__table__, properties = {
			'design': relation( Design,
				uselist = False ),
			'component': relation( Component,
				uselist = False )
			})

		class_mapper( Design ).add_property( '_components',
			relation( cls, collection_class = attribute_mapped_collection('component') ))

		Design.components = association_proxy('_components', 'amount', creator = lambda k,v: cls( component = k, amount = v ) )

	def InitMapper( cls, metadata, Component, Property ):
		cls.__table__ = Table( cls.__tablename__, metadata,
				Column('component_id', Integer, ForeignKey( Component.id ), primary_key = True ),
				Column('property_id',  Integer, ForeignKey( Property.id ), primary_key = True ),
				Column('value',        Text, nullable = False, default = """(lambda (design) 1)""" ))

		cols = cls.__table__.c

		Index('ix_%s_component_property' % cls.__tablename__, cols.component_id, cols.property_id)

		mapper( cls, cls.__table__, properties = {
			'component': relation( Component,
				uselist = False ),
			'property': relation( Property,
				uselist = False )
			})

		class_mapper( Component ).add_property( '_properties',
			relation( cls, collection_class = attribute_mapped_collection('property') ))

		Component.properties = association_proxy('_properties', 'value', creator = lambda k,v: cls( property = k, value = v ) )

    def __init__(self):
        try:
            from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
            from sqlalchemy.ext.declarative import declarative_base
            from sqlalchemy.orm import mapper, relationship
            from sqlalchemy.orm.collections import attribute_mapped_collection
        except ImportError:
            raise ImportError('Interaction with SQL based databases requires SQLAlchemy')

        self.metadata = MetaData()

        self.framework = Table('framework', self.metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(250)))

        self.argument = Table('argument', self.metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(250)),
            Column('framework_id', Integer, ForeignKey('framework.id')))

        self.attack= Table('attack', self.metadata,
            Column('id', Integer, primary_key=True),
            Column('attacker_id', Integer, ForeignKey('argument.id')),
            Column('target_id', Integer, ForeignKey('argument.id')))

        mapper(al.ArgumentationFramework, self.framework, properties={
            'argument' : relationship(al.Argument, backref='framework'),
            'arguments' : relationship(al.Argument,
                collection_class=attribute_mapped_collection('name'),
                cascade="all, delete-orphan")
        })

        mapper(al.Argument, self.argument, properties={
            'attacks' : relationship(al.Argument,
                secondary=self.attack,
                primaryjoin=self.argument.c.id==self.attack.c.attacker_id,
                secondaryjoin=self.argument.c.id==self.attack.c.target_id,
                collection_class=set)
        })

    def get_translations_relationship_args(self):
        foreign_keys = [
            getattr(self.translation_cls, column_key)
            for column_key in get_primary_keys(self.parent_cls).keys()
        ]

        relationship_args = copy(
            self.manager.option(
                self.parent_cls,
                'translations_relationship_args'
            )
        )
        defaults = dict(
            primaryjoin=sa.and_(*self.primary_key_conditions),
            foreign_keys=foreign_keys,
            collection_class=attribute_mapped_collection('locale'),
            comparator_factory=TranslationComparator,
            cascade='all, delete-orphan',
            passive_deletes=True,
        )
        for key, value in defaults.items():
            relationship_args.setdefault(key, value)
        return relationship_args

	def InitMapper( cls, metadata, Subject, Parameter, ParameterName ):
		subject = Subject.__origname__.lower()

		cls.__table__ = Table( cls.__tablename__, metadata,
				Column('%s_id' % subject, ForeignKey( Subject.id ), index = True, primary_key = True ),
				Column('name_id',  ForeignKey( ParameterName.id ), index = True, primary_key = True ),
				Column('param_id', ForeignKey( Parameter.id ), nullable = True ))

		mapper( cls, cls.__table__, properties = {
			'_name' : relation( ParameterName,
				uselist = False ),
			'parameter' : relation( Parameter,
				uselist = False )
			})

		cls.name = property( 
			lambda self: getattr( self._name, 'name', None ),
			lambda self, name: setattr( self, '_name', ParameterName.ByName(name) ) )

		class_mapper( Subject ).add_property( '_parameters',
			relation( cls, collection_class = attribute_mapped_collection('name') ))

		Subject.parameters = association_proxy('_parameters', 'parameter', creator = lambda k, v: cls( name = k, parameter = v ) )

def groupnode(klass, up='super', down='sub', name='node',
        keyattr='name', cascade="all, delete-orphan"):

    """
    Add Adjacency list attributes to klass.
    """

    uprel = up+name
    downrel = down+name+'s'
    idrel = up+name+'_id'

    _id = Column(Integer, ForeignKey('%s.id' % klass.__tablename__))
    remote_id = klass.metadata.tables[klass.__tablename__].columns['id']

    setattr(klass, idrel, _id)
    setattr(klass, downrel, relationship(klass,
        cascade=cascade,
        backref=backref(uprel, remote_side=[remote_id]),
        foreign_keys=[_id],
        collection_class=attribute_mapped_collection(keyattr),
    ))

    # Explicit indicator for root nodes, where remote_id is null
    setattr(klass, 'is_root'+name, Column(Boolean))

    def test_dict_collections(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        from sqlalchemy.orm.collections import attribute_mapped_collection

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        f = Foo()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr['hi'] = hi
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        f.someattr['there'] = there
        eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set([hi, there]), set(), set()))

        attributes.instance_state(f).commit(['someattr'])
        eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set(), set([hi, there]), set()))

#.........这里部分代码省略.........

    For Markdown-formatted columns, `(column)_map` and `(column)` will give
    Markdown objects.
    """
    # n.b.: language_class only exists for the sake of tests, which sometimes
    # want to create tables entirely separate from the pokedex metadata

    foreign_key_name = foreign_class.__singlename__ + '_id'

    Translations = type(_table_name, (object,), {
        '_language_identifier': association_proxy('local_language', 'identifier'),
        'relation_name': relation_name,
    })

    # Create the table object
    table = Table(_table_name, foreign_class.__table__.metadata,
        Column(foreign_key_name, Integer, ForeignKey(foreign_class.id),
            primary_key=True, nullable=False,
            info=dict(description="ID of the %s these texts relate to" % foreign_class.__singlename__)),
        Column('local_language_id', Integer, ForeignKey(language_class.id),
            primary_key=True, nullable=False,
            info=dict(description="Language these texts are in")),
    )
    Translations.__table__ = table

    # Add ye columns
    # Column objects have a _creation_order attribute in ascending order; use
    # this to get the (unordered) kwargs sorted correctly
    kwitems = kwargs.items()
    kwitems.sort(key=lambda kv: kv[1]._creation_order)
    for name, column in kwitems:
        column.name = name
        table.append_column(column)

    # Construct ye mapper
    mapper(Translations, table, properties={
        'foreign_id': synonym(foreign_key_name),
        'local_language': relationship(language_class,
            primaryjoin=table.c.local_language_id == language_class.id,
            innerjoin=True),
    })

    # Add full-table relations to the original class
    # Foo.bars_table
    setattr(foreign_class, relation_name + '_table', Translations)
    # Foo.bars
    setattr(foreign_class, relation_name, relationship(Translations,
        primaryjoin=foreign_class.id == Translations.foreign_id,
        collection_class=attribute_mapped_collection('local_language'),
    ))
    # Foo.bars_local
    # This is a bit clever; it uses bindparam() to make the join clause
    # modifiable on the fly.  db sessions know the current language and
    # populate the bindparam.
    # The 'dummy' value is to trick SQLA; without it, SQLA thinks this
    # bindparam is just its own auto-generated clause and everything gets
    # fucked up.
    local_relation_name = relation_name + '_local'
    setattr(foreign_class, local_relation_name, relationship(Translations,
        primaryjoin=and_(
            Translations.foreign_id == foreign_class.id,
            Translations.local_language_id == bindparam('_default_language_id',
                value='dummy', type_=Integer, required=True),
        ),
        foreign_keys=[Translations.foreign_id, Translations.local_language_id],
        uselist=False,
        #innerjoin=True,
        lazy=relation_lazy,
    ))

    # Add per-column proxies to the original class
    for name, column in kwitems:
        getset_factory = None
        string_getter = column.info.get('string_getter')
        if string_getter:
            getset_factory = _getset_factory_factory(
                column.name, string_getter)

        # Class.(column) -- accessor for the default language's value
        setattr(foreign_class, name,
            LocalAssociationProxy(local_relation_name, name,
                    getset_factory=getset_factory))

        # Class.(column)_map -- accessor for the language dict
        # Need a custom creator since Translations doesn't have an init, and
        # these are passed as *args anyway
        def creator(language, value):
            row = Translations()
            row.local_language = language
            setattr(row, name, value)
            return row
        setattr(foreign_class, name + '_map',
            association_proxy(relation_name, name, creator=creator,
                    getset_factory=getset_factory))

    # Add to the list of translation classes
    foreign_class.translation_classes.append(Translations)

    # Done
    return Translations