我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.orm.Query()。
def __init__(self, app=None, use_native_unicode=True, session_options=None, metadata=None): if session_options is None: session_options = {} session_options.setdefault('scopefunc', connection_stack.__ident_func__) self.use_native_unicode = use_native_unicode self.session = self.create_scoped_session(session_options) self.Model = self.make_declarative_base(metadata) self.Query = BaseQuery self._engine_lock = Lock() self.app = app _include_sqlalchemy(self) if app is not None: self.init_app(app)
def statement(self): """The full SELECT statement represented by this Query. The statement by default will not have disambiguating labels applied to the construct unless with_labels(True) is called first. """ stmt = self._compile_context(labels=self._with_labels).\ statement if self._params: stmt = stmt.params(self._params) # TODO: there's no tests covering effects of # the annotation not being there return stmt._annotate({'no_replacement_traverse': True})
def enable_eagerloads(self, value): """Control whether or not eager joins and subqueries are rendered. When set to False, the returned Query will not render eager joins regardless of :func:`~sqlalchemy.orm.joinedload`, :func:`~sqlalchemy.orm.subqueryload` options or mapper-level ``lazy='joined'``/``lazy='subquery'`` configurations. This is used primarily when nesting the Query's statement into a subquery or other selectable, or when using :meth:`.Query.yield_per`. """ self._enable_eagerloads = value
def with_labels(self): """Apply column labels to the return value of Query.statement. Indicates that this Query's `statement` accessor should return a SELECT statement that applies labels to all columns in the form <tablename>_<columnname>; this is commonly used to disambiguate columns from multiple tables which have the same name. When the `Query` actually issues SQL to load rows, it always uses column labeling. .. note:: The :meth:`.Query.with_labels` method *only* applies the output of :attr:`.Query.statement`, and *not* to any of the result-row invoking systems of :class:`.Query` itself, e.g. :meth:`.Query.first`, :meth:`.Query.all`, etc. To execute a query using :meth:`.Query.with_labels`, invoke the :attr:`.Query.statement` using :meth:`.Session.execute`:: result = session.execute(query.with_labels().statement) """ self._with_labels = True
def enable_assertions(self, value): """Control whether assertions are generated. When set to False, the returned Query will not assert its state before certain operations, including that LIMIT/OFFSET has not been applied when filter() is called, no criterion exists when get() is called, and no "from_statement()" exists when filter()/order_by()/group_by() etc. is called. This more permissive mode is used by custom Query subclasses to specify criterion or other modifiers outside of the usual usage patterns. Care should be taken to ensure that the usage pattern is even possible. A statement applied by from_statement() will override any criterion set by filter() or order_by(), for example. """ self._enable_assertions = value
def with_session(self, session): """Return a :class:`.Query` that will use the given :class:`.Session`. While the :class:`.Query` object is normally instantiated using the :meth:`.Session.query` method, it is legal to build the :class:`.Query` directly without necessarily using a :class:`.Session`. Such a :class:`.Query` object, or any :class:`.Query` already associated with a different :class:`.Session`, can produce a new :class:`.Query` object associated with a target session using this method:: from sqlalchemy.orm import Query query = Query([MyClass]).filter(MyClass.id == 5) result = query.with_session(my_session).one() """ self.session = session
def with_transformation(self, fn): """Return a new :class:`.Query` object transformed by the given function. E.g.:: def filter_something(criterion): def transform(q): return q.filter(criterion) return transform q = q.with_transformation(filter_something(x==5)) This allows ad-hoc recipes to be created for :class:`.Query` objects. See the example at :ref:`hybrid_transformers`. .. versionadded:: 0.7.4 """ return fn(self)
def with_hint(self, selectable, text, dialect_name='*'): """Add an indexing or other executional context hint for the given entity or selectable to this :class:`.Query`. Functionality is passed straight through to :meth:`~sqlalchemy.sql.expression.Select.with_hint`, with the addition that ``selectable`` can be a :class:`.Table`, :class:`.Alias`, or ORM entity / mapped class /etc. .. seealso:: :meth:`.Query.with_statement_hint` """ if selectable is not None: selectable = inspect(selectable).selectable self._with_hints += ((selectable, text, dialect_name),)
def group_by(self, *criterion): """apply one or more GROUP BY criterion to the query and return the newly resulting :class:`.Query` All existing GROUP BY settings can be suppressed by passing ``None`` - this will suppress any GROUP BY configured on mappers as well. .. versionadded:: 1.1 GROUP BY can be cancelled by passing None, in the same way as ORDER BY. """ if len(criterion) == 1: if criterion[0] is None: self._group_by = False return criterion = list(chain(*[_orm_columns(c) for c in criterion])) criterion = self._adapt_col_list(criterion) if self._group_by is False: self._group_by = criterion else: self._group_by = self._group_by + criterion
def outerjoin(self, *props, **kwargs): """Create a left outer join against this ``Query`` object's criterion and apply generatively, returning the newly resulting ``Query``. Usage is the same as the ``join()`` method. """ aliased, from_joinpoint, full = kwargs.pop('aliased', False), \ kwargs.pop('from_joinpoint', False), \ kwargs.pop('full', False) if kwargs: raise TypeError("unknown arguments: %s" % ', '.join(sorted(kwargs))) return self._join(props, outerjoin=True, full=full, create_aliases=aliased, from_joinpoint=from_joinpoint)
def suffix_with(self, *suffixes): r"""Apply the suffix to the query and return the newly resulting ``Query``. :param \*suffixes: optional suffixes, typically strings, not using any commas. .. versionadded:: 1.0.0 .. seealso:: :meth:`.Query.prefix_with` :meth:`.HasSuffixes.suffix_with` """ if self._suffixes: self._suffixes += suffixes else: self._suffixes = suffixes
def create_scoped_session(self, options=None): """Create a :class:`~sqlalchemy.orm.scoping.scoped_session` on the factory from :meth:`create_session`. An extra key ``'scopefunc'`` can be set on the ``options`` dict to specify a custom scope function. If it's not provided, Flask's app context stack identity is used. This will ensure that sessions are created and removed with the request/response cycle, and should be fine in most cases. :param options: dict of keyword arguments passed to session class in ``create_session`` """ if options is None: options = {} scopefunc = options.pop('scopefunc', _app_ctx_stack.__ident_func__) options.setdefault('query_cls', self.Query) return orm.scoped_session( self.create_session(options), scopefunc=scopefunc )
def __init__(self, app=None, use_native_unicode=True, session_options=None): self.use_native_unicode = use_native_unicode if session_options is None: session_options = {} session_options.setdefault( 'scopefunc', connection_stack.__ident_func__ ) self.session = self.create_scoped_session(session_options) self.Model = self.make_declarative_base() self._engine_lock = Lock() if app is not None: self.app = app self.init_app(app) else: self.app = None _include_sqlalchemy(self) _MapperSignalEvents(self.mapper).register() _SessionSignalEvents().register() self.Query = BaseQuery
def __init__(self, entities, session=None): """Construct a :class:`.Query` directly. E.g.:: q = Query([User, Address], session=some_session) The above is equivalent to:: q = some_session.query(User, Address) :param entities: a sequence of entities and/or SQL expressions. :param session: a :class:`.Session` with which the :class:`.Query` will be associated. Optional; a :class:`.Query` can be associated with a :class:`.Session` generatively via the :meth:`.Query.with_session` method as well. .. seealso:: :meth:`.Session.query` :meth:`.Query.with_session` """ self.session = session self._polymorphic_adapters = {} self._set_entities(entities)
def _only_entity_zero(self, rationale=None): if len(self._entities) > 1: raise sa_exc.InvalidRequestError( rationale or "This operation requires a Query " "against a single mapper." ) return self._entity_zero()
def _no_criterion_assertion(self, meth, order_by=True, distinct=True): if not self._enable_assertions: return if self._criterion is not None or \ self._statement is not None or self._from_obj or \ self._limit is not None or self._offset is not None or \ self._group_by or (order_by and self._order_by) or \ (distinct and self._distinct): raise sa_exc.InvalidRequestError( "Query.%s() being called on a " "Query with existing criterion. " % meth)
def _no_clauseelement_condition(self, meth): if not self._enable_assertions: return if self._order_by: raise sa_exc.InvalidRequestError( "Query.%s() being called on a " "Query with existing criterion. " % meth) self._no_criterion_condition(meth)
