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

Python url.make_url函数代码示例

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

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



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

示例1: test_comparison

    def test_comparison(self):
        components = (
            "drivername",
            "username",
            "password",
            "host",
            "database",
            "query",
            "port",
        )

        common_url = (
            "dbtype://username:password"
            "@[2001:da8:2004:1000:202:116:160:90]:80/database?foo=bar"
        )
        other_url = "dbtype://uname:[email protected]/"

        url1 = url.make_url(common_url)
        url2 = url.make_url(common_url)
        url3 = url.make_url(other_url)

        is_true(url1 == url2)
        is_false(url1 != url2)
        is_true(url1 != url3)
        is_false(url1 == url3)

        for curr_component in components:
            setattr(url2, curr_component, "new_changed_value")
            is_true(url1 != url2)
            is_false(url1 == url2)
            setattr(url2, curr_component, getattr(url1, curr_component))
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:31,代码来源:test_parseconnect.py


示例2: test_rfc1738_password

    def test_rfc1738_password(self):
        u = url.make_url("dbtype://user:pass word + other%[email protected]/dbname")
        eq_(u.password, "pass word + other:words")
        eq_(str(u), "dbtype://user:pass word + other%[email protected]/dbname")

        u = url.make_url(
            "dbtype://username:apples%[email protected]/database"
        )
        eq_(u.password, "apples/oranges")
        eq_(str(u), "dbtype://username:apples%[email protected]/database")

        u = url.make_url(
            "dbtype://username:apples%40oranges%40%[email protected]/database"
        )
        eq_(u.password, "[email protected]@@")
        eq_(
            str(u),
            "dbtype://username:apples%40oranges%40%[email protected]/database",
        )

        u = url.make_url("dbtype://username%40:@hostspec/database")
        eq_(u.password, "")
        eq_(u.username, "[email protected]")
        eq_(str(u), "dbtype://username%40:@hostspec/database")

        u = url.make_url("dbtype://username:pass%[email protected]/database")
        eq_(u.password, "pass/word")
        eq_(str(u), "dbtype://username:pass%[email protected]/database")
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:28,代码来源:test_parseconnect.py


示例3: _create_connection

def _create_connection(db):
    if isinstance(db, str):
        url = make_url(connections.databases[db]['ENGINE'])
    elif isinstance(db, dict):
        url = make_url(db['ENGINE'])
    else:
        url = db

    db_engine = url.drivername.split('+')[0]
    database = url.database
    kwargs = {'autocommit': True}

    if db_engine == 'sqlite':
        return
    elif db_engine == 'postgresql':
        database = 'postgres'
    elif db_engine == 'mssql':
        kwargs['isolation_level'] = 'AUTOCOMMIT'
        database = 'master'
    elif db_engine == 'mysql':
        database = None
    elif db_engine == 'oracle':
        database = connections.databases[DEFAULT_DB_ALIAS].get('SYSTEM_DB', 'SYSTEM')

    url = URL(url.drivername, url.username, url.password, url.host, url.port, database, url.query)
    return create_engine(url).connect().execution_options(**kwargs)
开发者ID:katrid,项目名称:orun,代码行数:26,代码来源:createdb.py


示例4: __init__

    def __init__(self):

        try:

            super(SQLRepository, self).__init__()

            self.conf = cfg.CONF
            url = None
            if self.conf.mysql.database_name is not None:
                settings_db = (self.conf.mysql.username,
                               self.conf.mysql.password,
                               self.conf.mysql.hostname,
                               self.conf.mysql.database_name)
                url = make_url("mysql+pymysql://%s:%[email protected]%s/%s" % settings_db)
            else:
                if self.conf.database.url is not None:
                    url = make_url(self.conf.database.url)
                else:
                    database_conf = dict(self.conf.database)
                    if 'url' in database_conf:
                        del database_conf['url']
                    url = URL(**database_conf)

            from sqlalchemy import create_engine
            self._db_engine = create_engine(url, pool_recycle=3600)

            self.metadata = MetaData()

        except Exception as ex:
            LOG.exception(ex)
            raise exceptions.RepositoryException(ex)
开发者ID:artur-ba,项目名称:monasca-api,代码行数:31,代码来源:sql_repository.py


示例5: test_pymssql_port_setting

    def test_pymssql_port_setting(self):
        dialect = pymssql.dialect()

        u = url.make_url("mssql+pymssql://scott:[email protected]/test")
        connection = dialect.create_connect_args(u)
        eq_(
            [
                [],
                {
                    "host": "somehost",
                    "password": "tiger",
                    "user": "scott",
                    "database": "test",
                },
            ],
            connection,
        )

        u = url.make_url("mssql+pymssql://scott:[email protected]:5000/test")
        connection = dialect.create_connect_args(u)
        eq_(
            [
                [],
                {
                    "host": "somehost:5000",
                    "password": "tiger",
                    "user": "scott",
                    "database": "test",
                },
            ],
            connection,
        )
开发者ID:monetate,项目名称:sqlalchemy,代码行数:32,代码来源:test_engine.py


示例6: configure

    def configure(cls, connection=None, url=None, dialect_name=None, opts={}):
        """Create a new :class:`.MigrationContext`.

        This is a factory method usually called
        by :meth:`.EnvironmentContext.configure`.

        :param connection: a :class:`~sqlalchemy.engine.Connection`
         to use for SQL execution in "online" mode.  When present,
         is also used to determine the type of dialect in use.
        :param url: a string database url, or a
         :class:`sqlalchemy.engine.url.URL` object.
         The type of dialect to be used will be derived from this if
         ``connection`` is not passed.
        :param dialect_name: string name of a dialect, such as
         "postgresql", "mssql", etc.  The type of dialect to be used will be
         derived from this if ``connection`` and ``url`` are not passed.
        :param opts: dictionary of options.  Most other options
         accepted by :meth:`.EnvironmentContext.configure` are passed via
         this dictionary.

        """
        if connection:
            dialect = connection.dialect
        elif url:
            url = sqla_url.make_url(url)
            dialect = url.get_dialect()()
        elif dialect_name:
            url = sqla_url.make_url("%s://" % dialect_name)
            dialect = url.get_dialect()()
        else:
            raise Exception("Connection, url, or dialect_name is required.")

        return MigrationContext(dialect, connection, opts)
开发者ID:leslyKay,项目名称:me,代码行数:33,代码来源:migration.py


示例7: all_to_intermediary

def all_to_intermediary(filename_or_input, schema=None):
    """ Dispatch the filename_or_input to the different function to produce the intermediary syntax.
    All the supported classes names are in `swich_input_class_to_method`.
    The input can also be a list of strings in markdown format or a filename finishing by '.er' containing markdown
    format.
    """
    # Try to convert from the name of the class
    input_class_name = filename_or_input.__class__.__name__
    try:
        this_to_intermediary = switch_input_class_to_method[input_class_name]
        tables, relationships = this_to_intermediary(filename_or_input)
        return tables, relationships
    except KeyError:
        pass

    # try to read markdown file.
    if isinstance(filename_or_input, basestring):
        if filename_or_input.split('.')[-1] == 'er':
            return markdown_file_to_intermediary(filename_or_input)

    # try to read a markdown in a string
    if not isinstance(filename_or_input, basestring):
        if all(isinstance(e, basestring) for e in filename_or_input):
            return line_iterator_to_intermediary(filename_or_input)

    # try to read DB URI.
    try:
        make_url(filename_or_input)
        return database_to_intermediary(filename_or_input, schema=schema)
    except ArgumentError:
        pass

    msg = 'Cannot process filename_or_input {}'.format(input_class_name)
    raise ValueError(msg)
开发者ID:jasondyjohnson,项目名称:eralchemy,代码行数:34,代码来源:main.py


示例8: normalize_url

def normalize_url(raw):
    """
    Build a URL from a string according to obvious rules.

    Converts *.sqlite paths to sqlite:/// URLs, etc.

    >>> str(normalize_url("/tmp/foo.sqlite"))
    'sqlite:////tmp/foo.sqlite'
    >>> str(normalize_url("postgresql://[email protected]/database"))
    'postgresql://[email protected]/database'
    """

    from sqlalchemy.engine.url import make_url
    from sqlalchemy.exceptions import ArgumentError

    try:
        return make_url(raw)
    except ArgumentError:
        from os.path import (
            abspath,
            splitext,
            )

        (root, extension) = splitext(raw)

        if extension == ".sqlite":
            return make_url("sqlite:///%s" % abspath(raw))
        else:
            raise ArgumentError("could not parse supposed URL \"%s\"" % raw)
开发者ID:borg-project,项目名称:cargo,代码行数:29,代码来源:alchemy.py


示例9: test_database_schema_hive

    def test_database_schema_hive(self):
        sqlalchemy_uri = 'hive://[email protected]:10000/default?auth=NOSASL'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)
        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('default', db)

        db = make_url(model.get_sqla_engine(schema='core_db').url).database
        self.assertEquals('core_db', db)
开发者ID:neuroradiology,项目名称:caravel,代码行数:8,代码来源:model_tests.py


示例10: test_database_schema_mysql

    def test_database_schema_mysql(self):
        sqlalchemy_uri = 'mysql://[email protected]/superset'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)

        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('superset', db)

        db = make_url(model.get_sqla_engine(schema='staging').url).database
        self.assertEquals('staging', db)
开发者ID:neuroradiology,项目名称:caravel,代码行数:9,代码来源:model_tests.py


示例11: test_database_schema_postgres

    def test_database_schema_postgres(self):
        sqlalchemy_uri = 'postgresql+psycopg2://postgres.airbnb.io:5439/prod'
        model = Database(sqlalchemy_uri=sqlalchemy_uri)

        db = make_url(model.get_sqla_engine().url).database
        self.assertEquals('prod', db)

        db = make_url(model.get_sqla_engine(schema='foo').url).database
        self.assertEquals('prod', db)
开发者ID:neuroradiology,项目名称:caravel,代码行数:9,代码来源:model_tests.py


示例12: get_absolute_uri_for_session

def get_absolute_uri_for_session(session):
    engine = session.get_bind()
    url_ = engine.url
    if url_.drivername.startswith('sqlite'):
        database_path = url_.database
        abs_database_path = os.path.abspath(database_path)
        abs_database_path = '/'.join(abs_database_path.split(os.sep))
        return make_url("sqlite:///%s" % abs_database_path)
    else:
        return make_url(url_)
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:10,代码来源:database_utils.py


示例13: test_plugin_multiple_url_registration

    def test_plugin_multiple_url_registration(self):
        from sqlalchemy.dialects import sqlite

        global MyEnginePlugin1
        global MyEnginePlugin2

        def side_effect_1(url, kw):
            eq_(kw, {"logging_name": "foob"})
            kw["logging_name"] = "bar"
            url.query.pop("myplugin1_arg", None)
            return MyEnginePlugin1

        def side_effect_2(url, kw):
            url.query.pop("myplugin2_arg", None)
            return MyEnginePlugin2

        MyEnginePlugin1 = Mock(side_effect=side_effect_1)
        MyEnginePlugin2 = Mock(side_effect=side_effect_2)

        plugins.register("engineplugin1", __name__, "MyEnginePlugin1")
        plugins.register("engineplugin2", __name__, "MyEnginePlugin2")

        e = create_engine(
            "sqlite:///?plugin=engineplugin1&foo=bar&myplugin1_arg=bat"
            "&plugin=engineplugin2&myplugin2_arg=hoho",
            logging_name="foob",
        )
        eq_(e.dialect.name, "sqlite")
        eq_(e.logging_name, "bar")

        # plugin args are removed from URL.
        eq_(e.url.query, {"foo": "bar"})
        assert isinstance(e.dialect, sqlite.dialect)

        eq_(
            MyEnginePlugin1.mock_calls,
            [
                call(url.make_url("sqlite:///?foo=bar"), {}),
                call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
                call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
                call.engine_created(e),
            ],
        )

        eq_(
            MyEnginePlugin2.mock_calls,
            [
                call(url.make_url("sqlite:///?foo=bar"), {}),
                call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
                call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
                call.engine_created(e),
            ],
        )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:53,代码来源:test_parseconnect.py


示例14: make_connection_manager

 def make_connection_manager(self, path):
     if isinstance(path, ConnectionManager):
         return path
     try:
         path = str(path)
         if path == "":
             return SQLiteConnectionManager("sqlite://")
         # Ensure that the path is in fact a URI
         make_url(path)
         return ConnectionManager(path)
     except:
         return self.connection_manager_type(path)
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:12,代码来源:connection.py


示例15: test_database_impersonate_user

    def test_database_impersonate_user(self):
        uri = 'mysql://[email protected]'
        example_user = 'giuseppe'
        model = Database(sqlalchemy_uri=uri)

        model.impersonate_user = True
        user_name = make_url(model.get_sqla_engine(user_name=example_user).url).username
        self.assertEquals(example_user, user_name)

        model.impersonate_user = False
        user_name = make_url(model.get_sqla_engine(user_name=example_user).url).username
        self.assertNotEquals(example_user, user_name)
开发者ID:neuroradiology,项目名称:caravel,代码行数:12,代码来源:model_tests.py


示例16: _assert_db_exists

def _assert_db_exists(engine):
    db_url = make_url(engine.url)
    db_name = db_url.database
    try:
        logger.debug('try to connect to database "%s": %s', db_name, db_url)
        connection = engine.connect()
        connection.close()
    except sqlalchemy.exc.OperationalError as err:
        db_url = make_url(engine.url)
        db_name = db_url.database
        logger.error('could not connect to database "%s": %s', db_name, str(err))
        raise ValueError('Cannot connect to database "%s".' % db_name)
开发者ID:dvischi,项目名称:TissueMAPS,代码行数:12,代码来源:utils.py


示例17: test_query_string

    def test_query_string(self):
        u = url.make_url("dialect://user:[email protected]/db?arg1=param1&arg2=param2")
        eq_(u.query, {"arg1": "param1", "arg2": "param2"})
        eq_(str(u), "dialect://user:[email protected]/db?arg1=param1&arg2=param2")

        u = url.make_url(
            "dialect://user:[email protected]/db?arg1=param1&arg2=param2&arg2=param3"
        )
        eq_(u.query, {"arg1": "param1", "arg2": ["param2", "param3"]})
        eq_(
            str(u),
            "dialect://user:[email protected]/db?arg1=param1&arg2=param2&arg2=param3",
        )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:13,代码来源:test_parseconnect.py


示例18: _url_from_target

 def _url_from_target(self, target):
     if isinstance(target, Connectable):
         return target.engine.url
     elif isinstance(target, six.string_types):
         if "://" not in target:
             target_url = sa_url.make_url("%s://" % target)
         else:
             target_url = sa_url.make_url(target)
         return target_url
     elif isinstance(target, sa_url.URL):
         return target
     else:
         raise ValueError("Invalid target type: %r" % target)
开发者ID:DinaBelova,项目名称:oslo.db,代码行数:13,代码来源:utils.py


示例19: check_db_url

def check_db_url(db_url):
    """Check for the correct format of the DB URL.  Return none if check
    passed.  Raise exception otherwise.
    :param db_url: The DB URL to be checked
    :raise exc.ArPrDataBadDBURLFormatException:
    :rtype : None
    """
    try:
        make_url(db_url)
    except Exception:
        raise exc.ArPrDataBadDBURLFormatException(
            "Incorrect DB URL Format.  Check http://docs.sqlalchemy.org"
            "/en/latest/core/engines.html?highlight=url#database-urls")
开发者ID:htquach,项目名称:ArPrData,代码行数:13,代码来源:db.py


示例20: __init__

    def __init__(self, url):
        self.table = Table('__tablename__', MetaData(),
                           Column('taskid', String(64), primary_key=True, nullable=False),
                           Column('project', String(64)),
                           Column('url', String(1024)),
                           Column('status', Integer),
                           Column('schedule', LargeBinary),
                           Column('fetch', LargeBinary),
                           Column('process', LargeBinary),
                           Column('track', LargeBinary),
                           Column('lastcrawltime', Float(32)),
                           Column('updatetime', Float(32)),
                           mysql_engine='InnoDB',
                           mysql_charset='utf8'
                           )

        self.url = make_url(url)
        if self.url.database:
            database = self.url.database
            self.url.database = None
            try:
                engine = create_engine(self.url, convert_unicode=True)
                engine.execute("CREATE DATABASE IF NOT EXISTS %s" % database)
            except sqlalchemy.exc.SQLAlchemyError:
                pass
            self.url.database = database
        self.engine = create_engine(self.url, convert_unicode=True)

        self._list_project()
开发者ID:bartqiao,项目名称:pyspider,代码行数:29,代码来源:taskdb.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python event.listen函数代码示例发布时间:2022-05-27
下一篇:
Python reflection.Inspector类代码示例发布时间: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