• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python util.coerce_kw_type函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中sqlalchemy.util.coerce_kw_type函数的典型用法代码示例。如果您正苦于以下问题:Python coerce_kw_type函数的具体用法?Python coerce_kw_type怎么用?Python coerce_kw_type使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了coerce_kw_type函数的18个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: create_connect_args

    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 "mode" in url.query:
            opts["mode"] = url.query["mode"]
            if isinstance(opts["mode"], basestring):
                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)
        # Can't set 'handle' or 'pool' via URL query args, use connect_args

        return ([], opts)
开发者ID:serah,项目名称:HR,代码行数:35,代码来源:oracle.py


示例2: create_connect_args

    def create_connect_args(self, url):
        if url.username or url.password or url.host or url.port:
            raise exc.ArgumentError(
                "Invalid SQLite URL: %s\n"
                "Valid SQLite URL forms are:\n"
                " sqlite:///:memory: (or, sqlite://)\n"
                " sqlite:///relative/path/to/file.db\n"
                " sqlite:////absolute/path/to/file.db" % (url,))
        filename = url.database or ':memory:'
        if filename != ':memory:':
            filename = os.path.abspath(filename)

        opts = self._driver_kwargs()
        opts.update(url.query)

        util.coerce_kw_type(opts, 'timeout', float)
        util.coerce_kw_type(opts, 'isolation_level', str)
        util.coerce_kw_type(opts, 'detect_types', int)
        util.coerce_kw_type(opts, 'check_same_thread', bool)
        util.coerce_kw_type(opts, 'cached_statements', int)

        jdbc_url = "jdbc:sqlite:%s" % filename
        return [
            [
                jdbc_url,
                url.username,
                url.password,
                self.jdbc_driver_name
            ],
            opts
        ]
开发者ID:parroit,项目名称:mod-alchemy-persistor,代码行数:31,代码来源:zxjdbc.py


示例3: create_connect_args

    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(
            threaded=self.threaded,
        )

        if dsn is not None:
            opts['dsn'] = dsn
        if url.password is not None:
            opts['password'] = url.password
        if url.username is not None:
            opts['user'] = url.username

        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)
开发者ID:eoghanmurray,项目名称:sqlalchemy,代码行数:55,代码来源:cx_oracle.py


示例4: create_connect_args

    def create_connect_args(self, url):
        filename = url.database or ':memory:'

        opts = url.query.copy()
        util.coerce_kw_type(opts, 'timeout', float)
        util.coerce_kw_type(opts, 'isolation_level', str)
        util.coerce_kw_type(opts, 'detect_types', int)
        util.coerce_kw_type(opts, 'check_same_thread', bool)
        util.coerce_kw_type(opts, 'cached_statements', int)

        return ([filename], opts)
开发者ID:BackupTheBerlios,项目名称:griffith-svn,代码行数:11,代码来源:sqlite.py


示例5: create_connect_args

    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,
            )

        # 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)
        # end Py2K

        if 'mode' in url.query:
            opts['mode'] = url.query['mode']
            if isinstance(opts['mode'], basestring):
                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)
开发者ID:NSkelsey,项目名称:research,代码行数:50,代码来源:cx_oracle.py


示例6: _coerce_config

def _coerce_config(configuration, prefix):
    """Convert configuration values to expected types."""

    options = dict([(key[len(prefix):], configuration[key])
                 for key in configuration if key.startswith(prefix)])
    for option, type_ in (
        ('convert_unicode', bool),
        ('pool_timeout', int),
        ('echo', bool),
        ('echo_pool', bool),
        ('pool_recycle', int),
        ('pool_size', int),
        ('max_overflow', int),
        ('pool_threadlocal', bool),
    ):
        util.coerce_kw_type(options, option, type_)
    return options
开发者ID:Eubolist,项目名称:ankimini,代码行数:17,代码来源:__init__.py


示例7: create_connect_args

    def create_connect_args(self, url):
        if url.username or url.password or url.host or url.port:
            raise exc.ArgumentError(
                "Invalid SQLite URL: %s\n"
                "Valid SQLite URL forms are:\n"
                " sqlite:///:memory: (or, sqlite://)\n"
                " sqlite:///relative/path/to/file.db\n"
                " sqlite:////absolute/path/to/file.db" % (url,))
        filename = url.database or ':memory:'

        opts = url.query.copy()
        util.coerce_kw_type(opts, 'timeout', float)
        util.coerce_kw_type(opts, 'isolation_level', str)
        util.coerce_kw_type(opts, 'detect_types', int)
        util.coerce_kw_type(opts, 'check_same_thread', bool)
        util.coerce_kw_type(opts, 'cached_statements', int)

        return ([filename], opts)
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:18,代码来源:pysqlite.py


示例8: create_connect_args

    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)
开发者ID:GabiR,项目名称:magazin_online,代码行数:44,代码来源:cx_oracle.py


示例9: _create_jdbc_url

	def _create_jdbc_url(self, url):
		"""Create a JDBC url from a :class:`~sqlalchemy.engine.url.URL`"""

		opts = url.translate_connect_args(database='db', username='user',password='passwd')
		opts.update(url.query)

		util.coerce_kw_type(opts, 'mode', str)

		print opts['mode']
		
		print url.database

		jdbc_url = 'jdbc:%s:%s://localhost/%s;' % (self.jdbc_db_name, opts['mode'], url.database)

		jdbc_url = ''

		print jdbc_url

		#MVCC=TRUE, can't change when db opened
		return jdbc_url
开发者ID:tliron,项目名称:sqlalchemy-jython,代码行数:20,代码来源:zxjdbc.py


示例10: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(username='user')
        opts.update(url.query)

        util.coerce_kw_type(opts, 'buffered', bool)
        util.coerce_kw_type(opts, 'raise_on_warnings', bool)
        opts['buffered'] = True
        opts['raise_on_warnings'] = True

        # FOUND_ROWS must be set in ClientFlag to enable
        # supports_sane_rowcount.
        if self.dbapi is not None:
            try:
                from mysql.connector.constants import ClientFlag
                client_flags = opts.get('client_flags', ClientFlag.get_default())
                client_flags |= ClientFlag.FOUND_ROWS
                opts['client_flags'] = client_flags
            except:
                pass
        return [[], opts]
开发者ID:Amelandbor,项目名称:CouchPotato,代码行数:20,代码来源:mysqlconnector.py


示例11: create_connect_args

 def create_connect_args(self, url):
     opts = url.translate_connect_args(username='user')
     if opts.get('port'):
         opts['host'] = "%s/%s" % (opts['host'], opts['port'])
         del opts['port']
     opts.update(url.query)
     
     util.coerce_kw_type(opts, 'type_conv', int)
     
     type_conv = opts.pop('type_conv', self.type_conv)
     concurrency_level = opts.pop('concurrency_level', self.concurrency_level)
     
     if self.dbapi is not None:
         initialized = getattr(self.dbapi, 'initialized', None)
         if initialized is None:
             # CVS rev 1.96 changed the name of the attribute:
             # http://kinterbasdb.cvs.sourceforge.net/viewvc/kinterbasdb/Kinterbasdb-3.0/__init__.py?r1=1.95&r2=1.96
             initialized = getattr(self.dbapi, '_initialized', False)
         if not initialized:
             self.dbapi.init(type_conv=type_conv, concurrency_level=concurrency_level)
     return ([], opts)
开发者ID:jsmiller84,项目名称:CouchPotato,代码行数:21,代码来源:kinterbasdb.py


示例12: create_connect_args

 def create_connect_args(self, url):
     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
         )
     opts.update(url.query)
     util.coerce_kw_type(opts, 'use_ansi', bool)
     return ([], opts)
开发者ID:BackupTheBerlios,项目名称:griffith-svn,代码行数:21,代码来源:oracle.py


示例13: create_connect_args

    def create_connect_args(self, url):
        if url.username or url.password:
            raise exc.ArgumentError(
                "Invalid RQLite URL: %s\n"
                "Valid RQLite URL forms are:\n"
                " rqlite+pyrqlite://host:port/[?params]" % (url,))

        opts = url.query.copy()
        util.coerce_kw_type(opts, 'connect_timeout', float)
        util.coerce_kw_type(opts, 'detect_types', int)
        util.coerce_kw_type(opts, 'max_redirects', int)
        opts['port'] = url.port
        opts['host'] = url.host

        return ([], opts)
开发者ID:rqlite,项目名称:sqlalchemy-rqlite,代码行数:15,代码来源:pyrqlite.py


示例14: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(database='db', username='user',
                                          password='passwd')
        opts.update(url.query)

        util.coerce_kw_type(opts, 'compress', bool)
        util.coerce_kw_type(opts, 'connect_timeout', int)
        util.coerce_kw_type(opts, 'client_flag', int)
        util.coerce_kw_type(opts, 'local_infile', int)
        # Note: using either of the below will cause all strings to be returned
        # as Unicode, both in raw SQL operations and with column types like
        # String and MSString.
        util.coerce_kw_type(opts, 'use_unicode', bool)
        util.coerce_kw_type(opts, 'charset', str)

        # Rich values 'cursorclass' and 'conv' are not supported via
        # query string.

        ssl = {}
        for key in ['ssl_ca', 'ssl_key', 'ssl_cert', 'ssl_capath', 'ssl_cipher']:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], str)
                del opts[key]
        if ssl:
            opts['ssl'] = ssl

        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
        # supports_sane_rowcount.
        client_flag = opts.get('client_flag', 0)
        if self.dbapi is not None:
            try:
                CLIENT_FLAGS = __import__(
                                    self.dbapi.__name__ + '.constants.CLIENT'
                                    ).constants.CLIENT
                client_flag |= CLIENT_FLAGS.FOUND_ROWS
            except (AttributeError, ImportError):
                pass
            opts['client_flag'] = client_flag
        return [[], opts]
开发者ID:denny820909,项目名称:builder,代码行数:40,代码来源:mysqldb.py


示例15: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(database="db", username="user", password="passwd")
        opts.update(url.query)

        util.coerce_kw_type(opts, "compress", bool)
        util.coerce_kw_type(opts, "connect_timeout", int)
        util.coerce_kw_type(opts, "client_flag", int)
        util.coerce_kw_type(opts, "local_infile", int)
        # Note: using either of the below will cause all strings to be returned
        # as Unicode, both in raw SQL operations and with column types like
        # String and MSString.
        util.coerce_kw_type(opts, "use_unicode", bool)
        util.coerce_kw_type(opts, "charset", str)

        # Rich values 'cursorclass' and 'conv' are not supported via
        # query string.

        ssl = {}
        for key in ["ssl_ca", "ssl_key", "ssl_cert", "ssl_capath", "ssl_cipher"]:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], str)
                del opts[key]
        if ssl:
            opts["ssl"] = ssl

        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
        # supports_sane_rowcount.
        client_flag = opts.get("client_flag", 0)
        if self.dbapi is not None:
            try:
                CLIENT_FLAGS = __import__(self.dbapi.__name__ + ".constants.CLIENT").constants.CLIENT
                client_flag |= CLIENT_FLAGS.FOUND_ROWS
            except (AttributeError, ImportError):
                self.supports_sane_rowcount = False
            opts["client_flag"] = client_flag
        return [[], opts]
开发者ID:swahtz,项目名称:evernote-classify,代码行数:37,代码来源:mysqldb.py


示例16: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(database='db', username='user',
                                          password='passwd')
        opts.update(url.query)

        util.coerce_kw_type(opts, 'port', int)
        util.coerce_kw_type(opts, 'compress', bool)
        util.coerce_kw_type(opts, 'autoping', bool)

        util.coerce_kw_type(opts, 'default_charset', bool)
        if opts.pop('default_charset', False):
            opts['charset'] = None
        else:
            util.coerce_kw_type(opts, 'charset', str)
        opts['use_unicode'] = opts.get('use_unicode', True)
        util.coerce_kw_type(opts, 'use_unicode', bool)

        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
        # supports_sane_rowcount.
        opts.setdefault('found_rows', True)

        ssl = {}
        for key in ['ssl_ca', 'ssl_key', 'ssl_cert', 
                        'ssl_capath', 'ssl_cipher']:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], str)
                del opts[key]
        if ssl:
            opts['ssl'] = ssl

        return [[], opts]
开发者ID:AntonNguyen,项目名称:easy_api,代码行数:32,代码来源:oursql.py


示例17: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(['host', 'db', 'user', 'passwd', 'port'])
        opts.update(url.query)

        util.coerce_kw_type(opts, 'compress', bool)
        util.coerce_kw_type(opts, 'connect_timeout', int)
        util.coerce_kw_type(opts, 'client_flag', int)
        util.coerce_kw_type(opts, 'local_infile', int)
        # note: these two could break SA Unicode type
        util.coerce_kw_type(opts, 'use_unicode', bool)   
        util.coerce_kw_type(opts, 'charset', str)
        # TODO: cursorclass and conv:  support via query string or punt?
        
        # ssl
        ssl = {}
        for key in ['ssl_ca', 'ssl_key', 'ssl_cert', 'ssl_capath', 'ssl_cipher']:
            if key in opts:
                ssl[key[4:]] = opts[key]
                util.coerce_kw_type(ssl, key[4:], str)
                del opts[key]
        if len(ssl):
            opts['ssl'] = ssl
        
        # FOUND_ROWS must be set in CLIENT_FLAGS for to enable
        # supports_sane_rowcount.
        client_flag = opts.get('client_flag', 0)
        if self.dbapi is not None:
            try:
                import MySQLdb.constants.CLIENT as CLIENT_FLAGS
                client_flag |= CLIENT_FLAGS.FOUND_ROWS
            except:
                pass
            opts['client_flag'] = client_flag
        return [[], opts]
开发者ID:BackupTheBerlios,项目名称:griffith-svn,代码行数:34,代码来源:mysql.py


示例18: create_connect_args

    def create_connect_args(self, url):
        opts = url.translate_connect_args(database='db', username='user',
                                          password='passwd')
        opts.update(url.query)

        util.coerce_kw_type(opts, 'port', int)
        util.coerce_kw_type(opts, 'compress', bool)
        util.coerce_kw_type(opts, 'autoping', bool)

        util.coerce_kw_type(opts, 'default_charset', bool)
        if opts.pop('default_charset', False):
            opts['charset'] = None
        else:
            util.coerce_kw_type(opts, 'charset', str)
        opts['use_unicode'] = opts.get('use_unicode', True)
        util.coerce_kw_type(opts, 'use_unicode', bool)

        # FOUND_ROWS must be set in CLIENT_FLAGS to enable
        # supports_sane_rowcount.
        opts.setdefault('found_rows', True)

        return [[], opts]
开发者ID:jsmiller84,项目名称:CouchPotato,代码行数:22,代码来源:oursql.py



注:本文中的sqlalchemy.util.coerce_kw_type函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python util.column_dict函数代码示例发布时间:2022-05-27
下一篇:
Python util.callable函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap