我们从Python开源项目中，提取了以下37个代码示例，用于说明如何使用sqlalchemy.util.string_types()。
def emits_warning_on(db, *warnings): """Mark a test as emitting a warning on a specific dialect. With no arguments, squelches all SAWarning failures. Or pass one or more strings; these will be matched to the root of the warning description by warnings.filterwarnings(). """ spec = db_spec(db) @decorator def decorate(fn, *args, **kw): if isinstance(db, util.string_types): if not spec(config._current): return fn(*args, **kw) else: wrapped = emits_warning(*warnings)(fn) return wrapped(*args, **kw) else: if not _is_excluded(*db): return fn(*args, **kw) else: wrapped = emits_warning(*warnings)(fn) return wrapped(*args, **kw) return decorate
def expect_warnings_on(db, *messages, **kw): """Context manager which expects one or more warnings on specific dialects. The expect version **asserts** that the warnings were in fact seen. """ spec = db_spec(db) if isinstance(db, util.string_types) and not spec(config._current): yield elif not _is_excluded(*db): yield else: with expect_warnings(*messages, **kw): yield
def expect_warnings_on(db, *messages, **kw): """Context manager which expects one or more warnings on specific dialects. The expect version **asserts** that the warnings were in fact seen. """ spec = db_spec(db) if isinstance(db, util.string_types) and not spec(config._current): yield else: with expect_warnings(*messages, **kw): yield
def create_connect_args(self, url): dialect_opts = dict(url.query) for opt in ('use_ansi', 'auto_setinputsizes', 'auto_convert_lobs', 'threaded', 'allow_twophase'): if opt in dialect_opts: util.coerce_kw_type(dialect_opts, opt, bool) setattr(self, opt, dialect_opts[opt]) database = url.database service_name = dialect_opts.get('service_name', None) if database or service_name: # if we have a database, then we have a remote host port = url.port if port: port = int(port) else: port = 1521 if database and service_name: raise exc.InvalidRequestError( '"service_name" option shouldn\'t ' 'be used with a "database" part of the url') if database: makedsn_kwargs = {'sid': database} if service_name: makedsn_kwargs = {'service_name': service_name} dsn = self.dbapi.makedsn(url.host, port, **makedsn_kwargs) else: # we have a local tnsname dsn = url.host opts = dict( user=url.username, password=url.password, dsn=dsn, threaded=self.threaded, twophase=self.allow_twophase, ) if util.py2k: if self._cx_oracle_with_unicode: for k, v in opts.items(): if isinstance(v, str): opts[k] = unicode(v) else: for k, v in opts.items(): if isinstance(v, unicode): opts[k] = str(v) if 'mode' in url.query: opts['mode'] = url.query['mode'] if isinstance(opts['mode'], util.string_types): mode = opts['mode'].upper() if mode == 'SYSDBA': opts['mode'] = self.dbapi.SYSDBA elif mode == 'SYSOPER': opts['mode'] = self.dbapi.SYSOPER else: util.coerce_kw_type(opts, 'mode', int) return ([], opts)
def create_connect_args(self, url): dialect_opts = dict(url.query) for opt in ('use_ansi', 'auto_setinputsizes', 'auto_convert_lobs', 'threaded', 'allow_twophase'): if opt in dialect_opts: util.coerce_kw_type(dialect_opts, opt, bool) setattr(self, opt, dialect_opts[opt]) if url.database: # if we have a database, then we have a remote host port = url.port if port: port = int(port) else: port = 1521 dsn = self.dbapi.makedsn(url.host, port, url.database) else: # we have a local tnsname dsn = url.host opts = dict( user=url.username, password=url.password, dsn=dsn, threaded=self.threaded, twophase=self.allow_twophase, ) if util.py2k: if self._cx_oracle_with_unicode: for k, v in opts.items(): if isinstance(v, str): opts[k] = unicode(v) else: for k, v in opts.items(): if isinstance(v, unicode): opts[k] = str(v) if 'mode' in url.query: opts['mode'] = url.query['mode'] if isinstance(opts['mode'], util.string_types): mode = opts['mode'].upper() if mode == 'SYSDBA': opts['mode'] = self.dbapi.SYSDBA elif mode == 'SYSOPER': opts['mode'] = self.dbapi.SYSOPER else: util.coerce_kw_type(opts, 'mode', int) return ([], opts)
