Python sqlalchemy.orm 模块，RelationshipProperty() 实例源码

def get_field_type(model, fieldname):
    """Helper which returns the SQLAlchemy type of the field.
    """
    field = getattr(model, fieldname)
    if isinstance(field, ColumnElement):
        fieldtype = field.type
    else:
        if isinstance(field, AssociationProxy):
            field = field.remote_attr
        if hasattr(field, 'property'):
            prop = field.property
            if isinstance(prop, RelProperty):
                return None
            fieldtype = prop.columns[0].type
        else:
            return None
    return fieldtype

def is_like_list(instance, relation):
    """Returns ``True`` if and only if the relation of `instance` whose name is
    `relation` is list-like.

    A relation may be like a list if, for example, it is a non-lazy one-to-many
    relation, or it is a dynamically loaded one-to-many.

    """
    if relation in instance._sa_class_manager:
        return instance._sa_class_manager[relation].property.uselist
    elif hasattr(instance, relation):
        attr = getattr(instance._sa_instance_state.class_, relation)
        if hasattr(attr, 'property'):
            return attr.property.uselist
    related_value = getattr(type(instance), relation, None)
    if isinstance(related_value, AssociationProxy):
        local_prop = related_value.local_attr.prop
        if isinstance(local_prop, RelProperty):
            return local_prop.uselist
    return False

def map(self, mode: EditMode, request: Request, context: t.Optional[Resource], model: type, includes: t.List, nested=None) -> colander.SchemaNode:
        """
        :param mode: What kind of form is this - show, add, edit
        :param request: HTTP request
        :param context: Current traversing context or None
        :param model: The SQLAlchemy model class for which we generate schema
        :param includes: List of column, relationship and property names or ``colander.SchemaNode(name=name)`` instances to be included on the form.
        :param nested: Legacy. Going away.
        """

        def _map_column(node, name, column, column_type):
            return self.map_column(mode, request, node, model, name, column, column_type)

        def _map_relationship(node: colander.SchemaNode, name: str, prop: RelationshipProperty, mapper: Mapper):
            return self.map_relationship(mode, request, node, model, name, prop, mapper)

        # TODO: We need to get rid of nested
        # Don't try to pull in relatinoships on add form
        nested = True

        # TODO: Make explicitly passed
        dbsession = request.dbsession

        schema = PropertyAwareSQLAlchemySchemaNode(model, dbsession, includes=includes, type_overrides=_map_column, relationship_overrides=_map_relationship, automatic_relationships=True, nested=nested)
        return schema

def _get_prop_value(self, clocked, prop):
        state = attributes.instance_state(clocked)

        # fires a load on any deferred columns
        if prop.key not in state.dict:
            getattr(clocked, prop.key)

        if isinstance(prop, orm.RelationshipProperty):
            changes = attributes.get_history(
                clocked, prop.key,
                passive=attributes.PASSIVE_NO_INITIALIZE)
        else:
            changes = attributes.get_history(clocked, prop.key)

        if changes.added:
            return changes.added[0]

        return NOT_FOUND_SENTINEL

def get_related_model(model, relationname):
    """Gets the class of the model to which `model` is related by the attribute
    whose name is `relationname`.

    """
    if hasattr(model, relationname):
        attr = getattr(model, relationname)
        if hasattr(attr, 'property') \
                and isinstance(attr.property, RelProperty):
            return attr.property.mapper.class_
        if isinstance(attr, AssociationProxy):
            return get_related_association_proxy_model(attr)
    return None

def db_matches_schema(Base, session):
    """
    Check whether the current database matches the models declared in model base.
    Currently we check that all tables exist with all columns.
    :param Base: Declarative Base for SQLAlchemy models to check
    :param session: SQLAlchemy session bound to an engine
    :return: True if all declared models have corresponding tables and columns.
    """

    engine = session.get_bind()
    iengine = inspect(engine)

    tables = iengine.get_table_names()

    # Go through all SQLAlchemy models
    for name, klass in Base._decl_class_registry.items():

        if isinstance(klass, _ModuleMarker):
            # Not a model
            continue

        table = klass.__table__.name
        if table in tables:

            columns = [c["name"] for c in iengine.get_columns(table)]
            mapper = inspect(klass)

            for column_prop in mapper.attrs:
                if isinstance(column_prop, RelationshipProperty):
                    # TODO: Add sanity checks for relations
                    pass
                else:
                    for column in column_prop.columns:
                        # Assume normal flat column
                        if column.key not in columns:
                            raise DBSchemaError("Model %s declares column %s which does not exist in database %s", klass, column.key, engine)
        else:
            raise DBSchemaError("Model %s declares table %s which does not exist in database %s", klass, table, engine)

def map_relationship(self, mode: EditMode, request: Request, node: colander.SchemaNode, model: type, name: str, rel: RelationshipProperty, mapper: Mapper):

        # Ok this is something we can handle, a single reference to another
        return self.map_standard_relationship(mode, request, node, model, name, rel)

def defaults_safety(*track, mapper):
    warnings.warn("these caveats are temporary", PendingDeprecationWarning)
    local_props = {mapper.get_property(prop) for prop in track}
    for prop in local_props:
        if isinstance(prop, orm.RelationshipProperty):
            continue
        assert all(col.onupdate is None for col in prop.columns), \
            '%r has onupdate' % prop
        assert all(col.server_default is None for col in prop.columns), \
            '%r has server_default' % prop
        assert all(col.server_onupdate is None for col in prop.columns), \
            '%r has server_onupdate' % prop


# TODO kwargs to override default clock table and history tables prefix

def build_history_class(
        cls: declarative.DeclarativeMeta,
        prop: T_PROPS,
        schema: str = None) -> nine.Type[TemporalProperty]:
    """build a sqlalchemy model for given prop"""
    class_name = "%s%s_%s" % (cls.__name__, 'History', prop.key)
    table = build_history_table(cls, prop, schema)
    base_classes = (
        TemporalProperty,
        declarative.declarative_base(metadata=table.metadata),
    )
    class_attrs = {
        '__table__': table,
        'entity': orm.relationship(
            lambda: cls,
            backref=orm.backref('%s_history' % prop.key, lazy='dynamic')
        ),
    }

    if isinstance(prop, orm.RelationshipProperty):
        class_attrs[prop.key] = orm.relationship(
            prop.argument,
            lazy='noload')

    model = type(class_name, base_classes, class_attrs)
    return model

def get_activity_clock_backref(
        activity: bases.TemporalActivityMixin,
        entity: bases.Clocked) -> orm.RelationshipProperty:
    """Get the backref'd clock history for a given entity."""
    assert (
        activity is entity.temporal_options.activity_cls or
        isinstance(activity, entity.temporal_options.activity_cls)
    ), "cannot inspect %r for mapped activity %r" % (entity, activity)

    inspected = sa.inspect(entity.temporal_options.activity_cls)
    backref = entity.temporal_options.clock_table.activity.property.backref

    return inspected.relationships[backref]

def relations(cls):
        """Return a `list` of relationship names or the given model
        """
        return [c.key for c in cls.__mapper__.iterate_properties
                if isinstance(c, RelationshipProperty)]

def prepare_for_export(cls, column, v):
        param_name = column.info.get('json') or to_camel_case(column.key)

        if hasattr(column, 'property') and isinstance(column.property, RelationshipProperty) and column.property.uselist:
            result = [c.to_dict() for c in v]

        elif hasattr(column, 'property') and isinstance(column.property, CompositeProperty):
            result = v.__composite_values__()

        elif v is None:
            result = v

        elif isinstance(v, datetime):
            result = format_iso_datetime(v)

        elif isinstance(v, date):
            result = v.isoformat()

        elif isinstance(v, time):
            result = format_iso_time(v)

        elif hasattr(v, 'to_dict'):
            result = v.to_dict()

        elif isinstance(v, Decimal):
            result = str(v)

        # Commented-out by vahid, I think it's not necessary at all.
        # elif isinstance(v, set):
        #     result = list(v)

        else:
            result = v

        return param_name, result

def is_sane_database(Base, session):
    """
    from: http://stackoverflow.com/questions/30428639/check-database-schema-matches-sqlalchemy-models-on-application-startup
    Check whether the current database matches the models declared in model base.

    Currently we check that all tables exist with all columns. What is not checked

    * Column types are not verified

    * Relationships are not verified at all (TODO)

    :param Base: Declarative Base for SQLAlchemy models to check

    :param session: SQLAlchemy session bound to an engine

    :return: True if all declared models have corresponding tables and columns.
    """
    engine = session.get_bind()
    iengine = inspect(engine)
    errors = False
    tables = iengine.get_table_names()

    # Go through all SQLAlchemy models
    for name, klass in Base._decl_class_registry.items():
        if isinstance(klass, _ModuleMarker):
            # Not a model
            continue
        table = klass.__tablename__
        if table in tables:
            # Check all columns are found
            # Looks like [{'default': "nextval('sanity_check_test_id_seq'::regclass)", 'autoincrement': True, 'nullable': False, 'type': INTEGER(), 'name': 'id'}]
            columns = [c["name"] for c in iengine.get_columns(table)]
            mapper = inspect(klass)
            for column_prop in mapper.attrs:
                if isinstance(column_prop, RelationshipProperty):
                    # TODO: Add sanity checks for relations
                    pass
                else:
                    for column in column_prop.columns:
                        # Assume normal flat column
                        if column.key not in columns:
                            logger.error(
                                "Model %s declares column %s which does not exist in database %s",
                                klass, column.key, engine
                            )
                            errors = True
        else:
            logger.error(
                "Model %s declares table %s which does not exist in database %s", klass, table,
                engine
            )
            errors = True
    return not errors

def map_standard_relationship(self, mode, request, node, model, name, rel) -> colander.SchemaNode:
        """Build a widget for choosing a relationship with target.

        The relationship must be foreign_key and the remote must offer ``uuid`` attribute which we use as a vocabulary key..
        """

        if isinstance(rel.argument, Mapper):
            # TODO: How to handle this kind of relationships
            return TypeOverridesHandling.drop

        if isinstance(rel.argument, _class_resolver):
            # Case from tutorialapp:
            # <RelationshipProperty at 0x1095a4438; question>
            # <class 'sqlalchemy.ext.declarative.clsregistry._class_resolver'>
            remote_model = rel.argument()
        else:
            remote_model = rel.argument

        # Get first column of the set
        for column in rel.local_columns:
            break

        # For now, we automatically deal with this only if the model provides uuid
        if hasattr(remote_model, "uuid"):
            dbsession = request.dbsession
            # TODO: We probably need a mechanism for system wide empty default label

            required = not column.nullable

            if mode in (EditMode.add, EditMode.edit):
                default_choice = "--- Choose one ---"
            else:
                default_choice = "(not set)"

            if required:
                missing = colander.required
            else:
                missing = None

            vocabulary = get_uuid_vocabulary_for_model(dbsession, remote_model, default_choice=default_choice)

            if rel.uselist:
                # Show out all relationships
                if mode == EditMode.show:
                    return colander.SchemaNode(UUIDModelSet(remote_model), name=name, missing=missing, widget=deform.widget.CheckboxChoiceWidget(values=vocabulary))
            else:
                # Select from a single relationship
                return colander.SchemaNode(UUIDForeignKeyValue(remote_model), name=name, missing=missing, widget=deform.widget.SelectWidget(values=vocabulary))

        return TypeOverridesHandling.drop

def build_history_table(
        cls: declarative.DeclarativeMeta,
        prop: T_PROPS,
        schema: str = None) -> sa.Table:
    """build a sql alchemy table for given prop"""
    if isinstance(prop, orm.RelationshipProperty):
        columns = [util.copy_column(column) for column in prop.local_columns]
    else:
        columns = [util.copy_column(column) for column in prop.columns]

    local_table = cls.__table__
    table_name = util.truncate_identifier(
        _generate_history_table_name(local_table, columns)
    )
    # Build the foreign key(s), specifically adding an index since we may use
    # a casted foreign key in our constraints. See _exclusion_in_uuid
    entity_foreign_keys = list(util.foreign_key_to(local_table, index=True))
    entity_constraints = [
        _exclusion_in(fk.type, fk.key)
        for fk in entity_foreign_keys
    ]

    constraints = [
        sa.Index(
            util.truncate_identifier('%s_effective_idx' % table_name),
            'effective',
            postgresql_using='gist'
        ),
        sap.ExcludeConstraint(
            *itertools.chain(entity_constraints, [('vclock', '&&')]),
            name=util.truncate_identifier('%s_excl_vclock' % table_name)
        ),
        sap.ExcludeConstraint(
            *itertools.chain(entity_constraints, [('effective', '&&')]),
            name=util.truncate_identifier('%s_excl_effective' % table_name)
        ),
    ]

    return sa.Table(
        table_name,
        local_table.metadata,
        sa.Column('id',
                  sap.UUID(as_uuid=True),
                  default=uuid.uuid4,
                  primary_key=True),
        sa.Column('effective',
                  sap.TSTZRANGE,
                  default=util.effective_now,
                  nullable=False),
        sa.Column('vclock', sap.INT4RANGE, nullable=False),
        *itertools.chain(entity_foreign_keys, columns, constraints),
        schema=schema or local_table.schema,
        keep_existing=True
    )  # memoization ftw

def is_sane_database(Base, session):
    """Check whether the current database matches the models declared in model base.
    Currently we check that all tables exist with all columns. What is not checked
    * Column types are not verified
    * Relationships are not verified at all (TODO)
    :param Base: Declarative Base for SQLAlchemy models to check
    :param session: SQLAlchemy session bound to an engine
    :return: True if all declared models have corresponding tables and columns.
    """

    engine = session.get_bind()
    iengine = inspect(engine)

    errors = False

    tables = iengine.get_table_names()

    # Go through all SQLAlchemy models
    for name, klass in Base._decl_class_registry.items():

        if isinstance(klass, _ModuleMarker):
            # Not a model
            continue

        table = klass.__tablename__
        if table in tables:
            # Check all columns are found
            # Looks like [{'default': "nextval('sanity_check_test_id_seq'::regclass)",
            # 'autoincrement': True, 'nullable': False, 'type': INTEGER(), 'name':
            # 'id'}]

            columns = [c["name"] for c in iengine.get_columns(table)]
            mapper = inspect(klass)

            for column_prop in mapper.attrs:
                if isinstance(column_prop, RelationshipProperty):
                    # TODO: Add sanity checks for relations
                    pass
                else:
                    for column in column_prop.columns:
                        # Assume normal flat column
                        if not column.key in columns:
                            logger.error(
                                "Model %s declares column %s which does not exist in database %s",
                                klass,
                                column.key,
                                engine)
                            errors = True
        else:
            logger.error("Model %s declares table %s which does not exist in database %s", klass, table, engine)
            errors = True

    return not errors