我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用sqlalchemy.pool.StaticPool()。
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 _setup_connection(self): """Ensure database is ready to fly.""" global Session # pylint: disable=global-statement import homeassistant.components.recorder.models as models from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session from sqlalchemy.orm import sessionmaker if self.db_url == 'sqlite://' or ':memory:' in self.db_url: from sqlalchemy.pool import StaticPool self.engine = create_engine( 'sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool) else: self.engine = create_engine(self.db_url, echo=False) models.Base.metadata.create_all(self.engine) session_factory = sessionmaker(bind=self.engine) Session = scoped_session(session_factory) self.db_ready.set()
def setup_bind(cls): if config.requirements.independent_connections.enabled: from sqlalchemy import pool return engines.testing_engine( options=dict(poolclass=pool.StaticPool)) else: return config.db
def _connect(self, url): self.url = url # TODO: seems like 0.5.x branch does not work with engine.dispose and staticpool #self.engine = create_engine(url, echo=True, poolclass=StaticPool) self.engine = create_engine(url, echo=True) # silence the logger added by SA, nose adds its own! logging.getLogger('sqlalchemy').handlers=[] self.meta = MetaData(bind=self.engine) if self.level < self.CONNECT: return #self.session = create_session(bind=self.engine) if self.level < self.TXN: return #self.txn = self.session.begin()
def create_database(): engine = create_engine('sqlite:///database.db', connect_args={'check_same_thread': False}, poolclass=StaticPool) Base.metadata.create_all(engine) return engine
def connect_database(): engine = create_engine('sqlite:///database.db', connect_args={'check_same_thread': False}, poolclass=StaticPool) return engine
def construct_engine(engine, **opts): """.. versionadded:: 0.5.4 Constructs and returns SQLAlchemy engine. Currently, there are 2 ways to pass create_engine options to :mod:`migrate.versioning.api` functions: :param engine: connection string or a existing engine :param engine_dict: python dictionary of options to pass to `create_engine` :param engine_arg_*: keyword parameters to pass to `create_engine` (evaluated with :func:`migrate.versioning.util.guess_obj_type`) :type engine_dict: dict :type engine: string or Engine instance :type engine_arg_*: string :returns: SQLAlchemy Engine .. note:: keyword parameters override ``engine_dict`` values. """ if isinstance(engine, Engine): return engine elif not isinstance(engine, basestring): raise ValueError("you need to pass either an existing engine or a database uri") # get options for create_engine if opts.get('engine_dict') and isinstance(opts['engine_dict'], dict): kwargs = opts['engine_dict'] else: kwargs = dict() # DEPRECATED: handle echo the old way echo = asbool(opts.get('echo', False)) if echo: warnings.warn('echo=True parameter is deprecated, pass ' 'engine_arg_echo=True or engine_dict={"echo": True}', exceptions.MigrateDeprecationWarning) kwargs['echo'] = echo # parse keyword arguments for key, value in opts.iteritems(): if key.startswith('engine_arg_'): kwargs[key[11:]] = guess_obj_type(value) log.debug('Constructing engine') # TODO: return create_engine(engine, poolclass=StaticPool, **kwargs) # seems like 0.5.x branch does not work with engine.dispose and staticpool return create_engine(engine, **kwargs)
def construct_engine(engine, **opts): """.. versionadded:: 0.5.4 Constructs and returns SQLAlchemy engine. Currently, there are 2 ways to pass create_engine options to :mod:`migrate.versioning.api` functions: :param engine: connection string or a existing engine :param engine_dict: python dictionary of options to pass to `create_engine` :param engine_arg_*: keyword parameters to pass to `create_engine` (evaluated with :func:`migrate.versioning.util.guess_obj_type`) :type engine_dict: dict :type engine: string or Engine instance :type engine_arg_*: string :returns: SQLAlchemy Engine .. note:: keyword parameters override ``engine_dict`` values. """ if isinstance(engine, Engine): return engine elif not isinstance(engine, six.string_types): raise ValueError("you need to pass either an existing engine or a database uri") # get options for create_engine if opts.get('engine_dict') and isinstance(opts['engine_dict'], dict): kwargs = opts['engine_dict'] else: kwargs = dict() # DEPRECATED: handle echo the old way echo = asbool(opts.get('echo', False)) if echo: warnings.warn('echo=True parameter is deprecated, pass ' 'engine_arg_echo=True or engine_dict={"echo": True}', exceptions.MigrateDeprecationWarning) kwargs['echo'] = echo # parse keyword arguments for key, value in six.iteritems(opts): if key.startswith('engine_arg_'): kwargs[key[11:]] = guess_obj_type(value) log.debug('Constructing engine') # TODO: return create_engine(engine, poolclass=StaticPool, **kwargs) # seems like 0.5.x branch does not work with engine.dispose and staticpool return create_engine(engine, **kwargs)
