我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.engine.Engine()。
def first_connect(self, dbapi_connection, connection_record): """Called exactly once for the first time a DBAPI connection is checked out from a particular :class:`.Pool`. The rationale for :meth:`.PoolEvents.first_connect` is to determine information about a particular series of database connections based on the settings used for all connections. Since a particular :class:`.Pool` refers to a single "creator" function (which in terms of a :class:`.Engine` refers to the URL and connection options used), it is typically valid to make observations about a single connection that can be safely assumed to be valid about all subsequent connections, such as the database version, the server and client encoding settings, collation settings, and many others. :param dbapi_connection: a DBAPI connection. :param connection_record: the :class:`._ConnectionRecord` managing the DBAPI connection. """
def set_engine_execution_options(self, engine, opts): """Intercept when the :meth:`.Engine.execution_options` method is called. The :meth:`.Engine.execution_options` method produces a shallow copy of the :class:`.Engine` which stores the new options. That new :class:`.Engine` is passed here. A particular application of this method is to add a :meth:`.ConnectionEvents.engine_connect` event handler to the given :class:`.Engine` which will perform some per- :class:`.Connection` task specific to these execution options. :param conn: The newly copied :class:`.Engine` object :param opts: dictionary of options that were passed to the :meth:`.Connection.execution_options` method. .. versionadded:: 0.9.0 .. seealso:: :meth:`.ConnectionEvents.set_connection_execution_options` - event which is called when :meth:`.Connection.execution_options` is called. """
def engine_disposed(self, engine): """Intercept when the :meth:`.Engine.dispose` method is called. The :meth:`.Engine.dispose` method instructs the engine to "dispose" of it's connection pool (e.g. :class:`.Pool`), and replaces it with a new one. Disposing of the old pool has the effect that existing checked-in connections are closed. The new pool does not establish any new connections until it is first used. This event can be used to indicate that resources related to the :class:`.Engine` should also be cleaned up, keeping in mind that the :class:`.Engine` can still be used for new requests in which case it re-acquires connection resources. .. versionadded:: 1.0.5 """
def init(config: dict, engine: Optional[Engine] = None) -> None: """ Initializes this module with the given config, registers all known command handlers and starts polling for message updates :param config: config to use :param engine: database engine for sqlalchemy (Optional) :return: None """ _CONF.update(config) if not engine: if _CONF.get('dry_run', False): engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool, echo=False) else: engine = create_engine('sqlite:///tradesv3.sqlite') session = scoped_session(sessionmaker(bind=engine, autoflush=True, autocommit=True)) Trade.session = session() Trade.query = session.query_property() _DECL_BASE.metadata.create_all(engine)
def test_traced_all_engines(self): # Don't register the engine explicitly. tracer = DummyTracer() sqlalchemy_opentracing.init_tracing(tracer, trace_all_engines=True, trace_all_queries=False) creat = CreateTable(self.users_table) sqlalchemy_opentracing.set_traced(creat) self.engine.execute(creat) # Unregister the main Engine class before doing our assertions, # in case we fail. sqlalchemy_opentracing.unregister_engine(Engine) self.assertEqual(1, len(tracer.spans)) self.assertEqual('create_table', tracer.spans[0].operation_name) self.assertEqual(True, tracer.spans[0].is_finished)
def init_tracing(tracer, trace_all_engines=True, trace_all_queries=True): ''' Set our global tracer. Tracer objects from our pyramid/flask/django libraries can be passed as well. ''' global g_tracer, g_trace_all_engines, g_trace_all_queries if hasattr(tracer, '_tracer'): tracer = tracer._tracer g_tracer = tracer g_trace_all_queries = trace_all_queries g_trace_all_engines = trace_all_engines if trace_all_engines: register_engine(Engine)
