我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.orm.mapper()。
def _accept_with(cls, orm, target): if isinstance(target, instrumentation.ClassManager): return target elif isinstance(target, mapperlib.Mapper): return target.class_manager elif target is orm.mapper: return instrumentation.ClassManager elif isinstance(target, type): if issubclass(target, mapperlib.Mapper): return instrumentation.ClassManager else: manager = instrumentation.manager_of_class(target) if manager: return manager else: return _InstanceEventsHold(target) return None
def instrument_class(self, mapper, class_): """Receive a class when the mapper is first constructed, before instrumentation is applied to the mapped class. This event is the earliest phase of mapper construction. Most attributes of the mapper are not yet initialized. This listener can either be applied to the :class:`.Mapper` class overall, or to any un-mapped class which serves as a base for classes that will be mapped (using the ``propagate=True`` flag):: Base = declarative_base() @event.listens_for(Base, "instrument_class", propagate=True) def on_new_class(mapper, cls_): " ... " :param mapper: the :class:`.Mapper` which is the target of this event. :param class\_: the mapped class. """
def reconstruct_instance(self, mapper, instance): """Receive an object instance after it has been created via ``__new__``, and after initial attribute population has occurred. This typically occurs when the instance is created based on incoming result rows, and is only called once for that instance's lifetime. Note that during a result-row load, this method is called upon the first row received for this instance. Note that some attributes and collections may or may not be loaded or even initialized, depending on what's present in the result rows. The return value is only significant within the ``MapperExtension`` chain; the parent mapper's behavior isn't modified by this method. """ return EXT_CONTINUE
def before_insert(self, mapper, connection, instance): """Receive an object instance before that instance is inserted into its table. This is a good place to set up primary key values and such that aren't handled otherwise. Column-based attributes can be modified within this method which will result in the new value being inserted. However *no* changes to the overall flush plan can be made, and manipulation of the ``Session`` will not have the desired effect. To manipulate the ``Session`` within an extension, use ``SessionExtension``. The return value is only significant within the ``MapperExtension`` chain; the parent mapper's behavior isn't modified by this method. """ return EXT_CONTINUE
def translate_row(self, mapper, context, row): """Perform pre-processing on the given result row and return a new row instance. This is called when the mapper first receives a row, before the object identity or the instance itself has been derived from that row. The given row may or may not be a ``RowProxy`` object - it will always be a dictionary-like object which contains mapped columns as keys. The returned object should also be a dictionary-like object which recognizes mapped columns as keys. If the ultimate return value is EXT_CONTINUE, the row is not translated. """ return EXT_CONTINUE
def create_instance(self, mapper, selectcontext, row, class_): """Receive a row when a new object instance is about to be created from that row. The method can choose to create the instance itself, or it can return EXT_CONTINUE to indicate normal object creation should take place. mapper The mapper doing the operation selectcontext The QueryContext generated from the Query. row The result row from the database class\_ The class we are mapping. return value A new object instance, or EXT_CONTINUE """ return EXT_CONTINUE
def populate_instance(self, mapper, selectcontext, row, instance, **flags): """Receive an instance before that instance has its attributes populated. This usually corresponds to a newly loaded instance but may also correspond to an already-loaded instance which has unloaded attributes to be populated. The method may be called many times for a single instance, as multiple result rows are used to populate eagerly loaded collections. If this method returns EXT_CONTINUE, instance population will proceed normally. If any other value or None is returned, instance population will not proceed, giving this extension an opportunity to populate the instance itself, if desired. .. deprecated:: 0.5 Most usages of this hook are obsolete. For a generic "object has been newly created from a row" hook, use ``reconstruct_instance()``, or the ``@orm.reconstructor`` decorator. """ return EXT_CONTINUE
def instrument_class(self, mapper, class_): r"""Receive a class when the mapper is first constructed, before instrumentation is applied to the mapped class. This event is the earliest phase of mapper construction. Most attributes of the mapper are not yet initialized. This listener can either be applied to the :class:`.Mapper` class overall, or to any un-mapped class which serves as a base for classes that will be mapped (using the ``propagate=True`` flag):: Base = declarative_base() @event.listens_for(Base, "instrument_class", propagate=True) def on_new_class(mapper, cls_): " ... " :param mapper: the :class:`.Mapper` which is the target of this event. :param class\_: the mapped class. """
def loaded_as_persistent(self, session, instance): """Intercept the "loaded as persistent" transition for a specific object. This event is invoked within the ORM loading process, and is invoked very similarly to the :meth:`.InstanceEvents.load` event. However, the event here is linkable to a :class:`.Session` class or instance, rather than to a mapper or class hierarchy, and integrates with the other session lifecycle events smoothly. The object is guaranteed to be present in the session's identity map when this event is called. :param session: target :class:`.Session` :param instance: the ORM-mapped instance being operated upon. .. versionadded:: 1.1 .. seealso:: :ref:`session_lifecycle_events` """
def _commit(self, rows): if not self._parameters.get("only-save-schema"): self.db_connect(retry=True) mapper = self._get_mapper() update_rows = [] insert_rows = [] for row in rows: if not self._is_row_in(row, update_rows + insert_rows): filter_args = (getattr(self.db_table.c, field)==row[field] for field in self._update_keys) if self.db_session.query(self.db_table).filter(*filter_args).count() > 0: update_rows.append(row) else: insert_rows.append(row) if len(insert_rows) > 0: self.db_session.bulk_insert_mappings(mapper, insert_rows) self._incr_stat("inserted rows", len(insert_rows)) if len(update_rows) > 0: self.db_session.bulk_update_mappings(mapper, update_rows) self._incr_stat("updated rows", len(update_rows)) self.db_commit() logging.info("{}: commit ({} updated, {} inserted)".format(self._log_prefix, self._get_stat("updated rows", 0), self._get_stat("inserted rows", 0))) # force a new session on next commit self._db_session = None
def import_upon_configure(Base, here): """ Import ORM models from the given base class into the current namespace. """ def import_models_into_namespace(): for class_ in Base._decl_class_registry.values(): if hasattr(class_, '__tablename__'): setattr(here, class_.__name__, class_) # Listen for the SQLAlchemy event and run setup_schema. # Note: This has to be done after Base and session are setup event.listen(mapper, 'after_configured', import_models_into_namespace)
def _new_mapper_instance(cls, class_, mapper): _MapperEventsHold.populate(class_, mapper)
def _accept_with(cls, orm, target): if target is orm.mapper: return mapperlib.Mapper elif isinstance(target, type): if issubclass(target, mapperlib.Mapper): return target else: mapper = _mapper_or_none(target) if mapper is not None: return mapper else: return _MapperEventsHold(target) else: return target
def _listen( cls, event_key, raw=False, retval=False, propagate=False, **kw): target, identifier, fn = \ event_key.dispatch_target, event_key.identifier, \ event_key._listen_fn if identifier in ("before_configured", "after_configured") and \ target is not mapperlib.Mapper: util.warn( "'before_configured' and 'after_configured' ORM events " "only invoke with the mapper() function or Mapper class " "as the target.") if not raw or not retval: if not raw: meth = getattr(cls, identifier) try: target_index = \ inspect_getargspec(meth)[0].index('target') - 1 except ValueError: target_index = None def wrap(*arg, **kw): if not raw and target_index is not None: arg = list(arg) arg[target_index] = arg[target_index].obj() if not retval: fn(*arg, **kw) return interfaces.EXT_CONTINUE else: return fn(*arg, **kw) event_key = event_key.with_wrapper(wrap) if propagate: for mapper in target.self_and_descendants: event_key.with_dispatch_target(mapper).base_listen( propagate=True, **kw) else: event_key.base_listen(**kw)
