我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.pool.NullPool()。
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ connectable = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata ) with context.begin_transaction(): context.run_migrations()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) connection = engine.connect() context.configure( connection=connection, target_metadata=target_metadata ) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def _load_db_pool_config(self, config): """????????? """ pool_config = {} for section in config.prefix("dbpool_"): pool_name = section.split("_", 1)[1] pool_config[pool_name] = config[section] # ???????????????? poolclass = pool_config[pool_name]["poolclass"] if isinstance(poolclass, str): if poolclass.lower() == "queuepool": poolclass = QueuePool elif poolclass.lower() == "nullpool": poolclass = NullPool else: module_name, class_name = poolclass.rsplit(".", 1) poolclass = getattr(__import__(module_name), class_name) pool_config[pool_name]["poolclass"] = poolclass for item_name in ("pool_size", "pool_recycle", "pool_timeout"): item_value = pool_config[pool_name][item_name] pool_config[pool_name][item_name] = int(item_value) return pool_config
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ connectable = create_engine( url, poolclass=pool.NullPool ) with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, version_table='results_schema_versions', include_schemas=True, ) connection.execute('set search_path to "{}", public'.format('results')) with context.begin_transaction(): context.run_migrations()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ db_url = os.environ['DRYDOCK_DB_URL'] connectable = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool, url=db_url) with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata) with context.begin_transaction(): context.run_migrations()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool ) connection = engine.connect() context.configure(connection=connection, target_metadata=target_metadata) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = engine_from_config(config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) connection = engine.connect() context.configure(connection=connection, target_metadata=target_metadata) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = create_engine( qinling_config.database.connection, poolclass=pool.NullPool ) connection = engine.connect() context.configure( connection=connection, target_metadata=target_metadata ) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) connection = engine.connect() context.configure( connection=connection, target_metadata=target_metadata, version_table=config.get_main_option('version_table')) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ connectable = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, version_table="un_alembic_version", include_object=include_object, ) with context.begin_transaction(): context.run_migrations()
def get_engine(connection_string): """ Get a SQLAlchemy engine that allows us to connect to a database. """ # Set echo to False, as otherwise we get full SQL Query outputted, which can overwhelm the terminal engine = create_engine( connection_string, echo=False, connect_args={'check_same_thread': False}, poolclass=NullPool, convert_unicode=True, ) if connection_string == get_default_db_string() and connection_string.startswith('sqlite'): event.listen(engine, "connect", set_sqlite_connection_pragma) connection = engine.connect() connection.execute(START_PRAGMAS) connection.close() return engine
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ url = app.config['SQLALCHEMY_DATABASE_URI'] engine = create_engine(url, poolclass=pool.NullPool) connection = engine.connect() context.configure( connection=connection, target_metadata=target_metadata, include_symbol=include_symbol ) try: with context.begin_transaction(): connection.execute('SELECT pg_advisory_lock(1)') context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = create_engine( neutron_config.database.connection, poolclass=pool.NullPool) connection = engine.connect() context.configure( connection=connection, target_metadata=target_metadata) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def run_migrations_online(): """Run migrations in 'online' mode. In this scenario we need to create an Engine and associate a connection with the context. """ engine = engine_from_config(config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) connection = engine.connect() context.configure(connection=connection, target_metadata=target_metadata, **current_app.extensions['migrate'].configure_args) try: with context.begin_transaction(): context.run_migrations() finally: connection.close()
def reflect_hints_db(db_path): """ Reflect the database schema of the hints database, automapping the existing tables The NullPool is used to avoid concurrency issues with luigi. Using this activates pooling, but since sqlite doesn't really support pooling, what effectively happens is just that it locks the database and the other connections wait. :param db_path: path to hints sqlite database :return: sqlalchemy.MetaData object, sqlalchemy.orm.Session object """ engine = sqlalchemy.create_engine('sqlite:///{}'.format(db_path), poolclass=NullPool) metadata = sqlalchemy.MetaData() metadata.reflect(bind=engine) Base = automap_base(metadata=metadata) Base.prepare() speciesnames = Base.classes.speciesnames seqnames = Base.classes.seqnames hints = Base.classes.hints featuretypes = Base.classes.featuretypes Session = sessionmaker(bind=engine) session = Session() return speciesnames, seqnames, hints, featuretypes, session
