我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.engine.url.URL。
def create(self, name_or_url, executor, **kwargs): # create url.URL object u = url.make_url(name_or_url) dialect_cls = u.get_dialect() dialect_args = {} # consume dialect arguments from kwargs for k in util.get_cls_kwargs(dialect_cls): if k in kwargs: dialect_args[k] = kwargs.pop(k) # create dialect dialect = dialect_cls(**dialect_args) return MockEngineStrategy.MockConnection(dialect, executor)
def get_url(self): dbconf=self.dbconf try: dialect=get_item(dbconf, 'dialect', ) except Exception as e: raise Exception("dialect key missing in configuration for: %s" % (self.database)) from e func=dialect_map.get(dialect) if func: #except Exception as e: #raise Exception('unhandled dialect: %s, allow %s.' %(dialect, repr(list(dialect_map.keys())))) from e logger.debug('specialize map: %s' % repr(func)) url, kwargs=func(dbconf, self.database) else: url_args=get_url_kwargs(dbconf, ) try: url=URL(drivername=dialect, **url_args) # fetch other connection args (called query in sqlalchemy.engine.url.URL except Exception as e: raise Exception('Failed to create URL for : %s:%s' % (dialect, url_args.get('drivername'))) from e kwargs=get_rest_kwargs(dbconf, ignore=CONNECT_ARGS, ) return url, kwargs
def run_migrations_offline(): """Run migrations in 'offline' mode. This configures the context with just a URL and not an Engine, though an Engine is acceptable here as well. By skipping the Engine creation we don't even need a DBAPI to be available. Calls to context.execute() here emit the given string to the script output. """ context.configure( url=url, target_metadata=target_metadata, literal_binds=True, version_table='results_schema_versions' ) with context.begin_transaction(): context.run_migrations()
def __call__(self, cfg, *arg): if isinstance(cfg, compat.string_types): url = sa_url.make_url(cfg) elif isinstance(cfg, sa_url.URL): url = cfg else: url = cfg.db.url backend = url.get_backend_name() if backend in self.fns: return self.fns[backend](cfg, *arg) else: return self.fns['*'](cfg, *arg)
def _compile_dialect(self, execute_observed): if self.dialect == 'default': return DefaultDialect() else: # ugh if self.dialect == 'postgresql': params = {'implicit_returning': True} else: params = {} return url.URL(self.dialect).get_dialect()(**params)
def __call__(self, cfg, *arg): if isinstance(cfg, compat.string_types): url = sa_url.make_url(cfg) elif isinstance(cfg, sa_url.URL): url = cfg else: url = cfg.db.url backend = get_url_backend_name(url) if backend in self.fns: return self.fns[backend](cfg, *arg) else: return self.fns['*'](cfg, *arg)
def init_db(db_type, username, password, host, db, db_location): """ Creates database engine :param db_type: database type :param username: database username :param password: database password :param host: database host :param db: database name :return: """ global engine url = None if db_type == 'mysql': url = URL(db_type, username, password, host, database=db) elif db_type == 'sqlite': url = "sqlite:///" + db_location engine = create_engine(url)
def sqlite_url(dbconf, database): ''' engine = create_engine('sqlite:///foo.db') engine = create_engine('sqlite:////absolute/path/to/foo.db') engine = create_engine('sqlite:///C:\\path\\to\\foo.db') engine = create_engine(r'sqlite:///C:\path\to\foo.db') engine = create_engine('sqlite://') ''' creator=None file=dbconf.get('database') url_args=get_url_kwargs(dbconf, ) url=URL(drivername='sqlite', **url_args) if not file: PY2 = sys.version_info.major == 2 if PY2: params = {} else: params = {'uri': True} url_args_list=["%s=%s" % (k, v) for k, v in url_args.items()] url_args_str='' if url_args_list: url_args_str="?%s" % '&'.join(url_args_list) DB_URI = 'file::memory:%s' %(url_args_str) creator = lambda: sqlite3.connect(DB_URI, **params) kwargs=get_rest_kwargs(dbconf, ignore=CONNECT_ARGS, ) if creator: kwargs['creator']=creator return url, kwargs
def connect(poolclass=QueuePool): with open('database.yaml') as fd: config = yaml.load(fd) dburl = URL( 'postgres', host=config['host'], username=config['user'], database=config['db'], password=config['pass'], port=config['port'], ) return create_engine(dburl, poolclass=poolclass)
