SELECT address.email_address, user.name FROM user JOIN address ON user.id == address.user_id
+.. _change_5526:
+
+The URL object is now immutable
+-------------------------------
+
+The :class:`_engine.URL` object has been formalized such that it now presents
+itself as a ``namedtuple`` with a fixed number of fields that are immutable. In
+addition, the dictionary represented by the :attr:`_engine.URL.query` attribute
+is also an immutable mapping. Mutation of the :class:`_engine.URL` object was
+not a formally supported or documented use case which led to some open-ended
+use cases that made it very difficult to intercept incorrect usages, most
+commonly mutation of the :attr:`_engine.URL.query` dictionary to include non-string elements.
+It also led to all the common problems of allowing mutability in a fundamental
+data object, namely unwanted mutations elsewhere leaking into code that didn't
+expect the URL to change. Finally, the namedtuple design is inspired by that
+of Python's ``urllib.parse.urlparse()`` which returns the parsed object as a
+named tuple.
+
+The decision to change the API outright is based on a calculus weighing the
+infeasability of a deprecation path (which would involve changing the
+:attr:`_engine.URL.query` dictionary to be a special dictionary that emits deprecation
+warnings when any kind of standard library mutation methods are invoked, in
+addition that when the dictionary would hold any kind of list of elements, the
+list would also have to emit deprecation warnings on mutation) against the
+unlikely use case of projects already mutating :class:`_engine.URL` objects in
+the first place, as well as that small changes such as that of :ticket:`5341`
+were creating backwards-incompatibility in any case. The primary case for
+mutation of a
+:class:`_engine.URL` object is that of parsing plugin arguments within the
+:class:`_engine.CreateEnginePlugin` extension point, itself a fairly recent
+addition that based on Github code search is in use by two repositories,
+neither of which are actually mutating the URL object.
+
+The :class:`_engine.URL` object now provides a rich interface inspecting
+and generating new :class:`_engine.URL` objects. The
+existing mechanism to create a :class:`_engine.URL` object, the
+:func:`_engine.make_url` function, remains unchanged::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
+
+For programmatic construction, code that may have been using the
+:class:`_engine.URL` constructor or ``__init__`` method directly will
+receive a deprecation warning if arguments are passed as keyword arguments
+and not an exact 7-tuple. The keyword-style constructor is now available
+via the :meth:`_engine.URL.create` method::
+
+ >>> from sqlalchemy.engine import URL
+ >>> url = URL.create("postgresql", "user", "pass", host="host", database="dbname")
+ >>> str(url)
+ 'postgresql://user:pass@host/dbname'
+
+
+Fields can be altered typically using the :meth:`_engine.URL.set` method, which
+returns a new :class:`_engine.URL` object with changes applied::
+
+ >>> mysql_url = url.set(drivername="mysql+pymysql")
+ >>> str(mysql_url)
+ 'mysql+pymysql://user:pass@host/dbname'
+
+To alter the contents of the :attr:`_engine.URL.query` dictionary, methods
+such as :meth:`_engine.URL.update_query_dict` may be used::
+
+ >>> url.update_query_dict({"sslcert": '/path/to/crt'})
+ postgresql://user:***@host/dbname?sslcert=%2Fpath%2Fto%2Fcrt
+
+To upgrade code that is mutating these fields directly, a **backwards and
+forwards compatible approach** is to use a duck-typing, as in the following
+style::
+
+ def set_url_drivername(some_url, some_drivername):
+ # check for 1.4
+ if hasattr(some_url, "set"):
+ return some_url.set(drivername=some_drivername)
+ else:
+ # SQLAlchemy 1.3 or earlier, mutate in place
+ some_url.drivername = some_drivername
+ return some_url
+
+ def set_ssl_cert(some_url, ssl_cert):
+ # check for 1.4
+ if hasattr(some_url, "update_query_dict"):
+ return some_url.update_query_dict({"sslcert": ssl_cert})
+ else:
+ # SQLAlchemy 1.3 or earlier, mutate in place
+ some_url.query["sslcert"] = ssl_cert
+ return some_url
+
+The query string retains its existing format as a dictionary of strings
+to strings, using sequences of strings to represent multiple parameters.
+For example::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&sslcert=%2Fpath%2Fto%2Fcrt")
+ >>> url.query
+ immutabledict({'alt_host': ('host1', 'host2'), 'sslcert': '/path/to/crt'})
+
+To work with the contents of the :attr:`_engine.URL.query` attribute such that all values are
+normalized into sequences, use the :attr:`_engine.URL.normalized_query` attribute::
+
+ >>> url.normalized_query
+ immutabledict({'alt_host': ('host1', 'host2'), 'sslcert': ('/path/to/crt',)})
+
+The query string can be appended to via methods such as :meth:`_engine.URL.update_query_dict`,
+:meth:`_engine.URL.update_query_pairs`, :meth:`_engine.URL.update_query_string`::
+
+ >>> url.update_query_dict({"alt_host": "host3"}, append=True)
+ postgresql://user:***@host/dbname?alt_host=host1&alt_host=host2&alt_host=host3&sslcert=%2Fpath%2Fto%2Fcrt
+
+.. seealso::
+
+ :class:`_engine.URL`
+
+
+Changes to CreateEnginePlugin
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The :class:`_engine.CreateEnginePlugin` is also impacted by this change,
+as the documentation for custom plugins indicated that the ``dict.pop()``
+method should be used to remove consumed arguments from the URL object. This
+should now be acheived using the :meth:`_engine.CreateEnginePlugin.update_url`
+method. A backwards compatible approach would look like::
+
+ from sqlalchemy.engine import CreateEnginePlugin
+
+ class MyPlugin(CreateEnginePlugin):
+ def __init__(self, url, kwargs):
+ # check for 1.4 style
+ if hasattr(CreateEnginePlugin, "update_url"):
+ self.my_argument_one = url.query['my_argument_one']
+ self.my_argument_two = url.query['my_argument_two']
+ else:
+ # legacy
+ self.my_argument_one = url.query.pop('my_argument_one')
+ self.my_argument_two = url.query.pop('my_argument_two')
+
+ self.my_argument_three = kwargs.pop('my_argument_three', None)
+
+ def update_url(self, url):
+ # this method runs in 1.4 only and should be used to consume
+ # plugin-specific arguments
+ return url.difference_update_query(
+ ["my_argument_one", "my_argument_two"]
+ )
+
+See the docstring at :class:`_engine.CreateEnginePlugin` for complete details
+on how this class is used.
+
+:ticket:`5526`
+
+
.. _change_5284:
select(), case() now accept positional expressions
--- /dev/null
+.. change::
+ :tags: change, engine
+ :tickets: 5526
+
+ The :class:`_engine.URL` object is now an immutable named tuple. To modify
+ a URL object, use the :meth:`_engine.URL.set` method to produce a new URL
+ object.
+
+ .. seealso::
+
+ :ref:`change_5526` - notes on migration
+
"sqlalchemy.sql.base": "sqlalchemy.sql.expression",
"sqlalchemy.event.base": "sqlalchemy.event",
"sqlalchemy.engine.base": "sqlalchemy.engine",
+ "sqlalchemy.engine.url": "sqlalchemy.engine",
"sqlalchemy.engine.row": "sqlalchemy.engine",
"sqlalchemy.engine.cursor": "sqlalchemy.engine",
"sqlalchemy.engine.result": "sqlalchemy.engine",
zzzeeksphinx_module_prefixes = {
"_sa": "sqlalchemy",
"_engine": "sqlalchemy.engine",
+ "_url": "sqlalchemy.engine",
"_result": "sqlalchemy.engine",
"_row": "sqlalchemy.engine",
"_schema": "sqlalchemy.schema",
.. autofunction:: sqlalchemy.create_mock_engine
-.. autofunction:: sqlalchemy.engine.url.make_url
+.. autofunction:: sqlalchemy.engine.make_url
-.. autoclass:: sqlalchemy.engine.url.URL
+.. autoclass:: sqlalchemy.engine.URL
:members:
+ .. py:attribute:: drivername
+ :annotation: str
+
+ database backend and driver name, such as
+ ``postgresql+psycopg2``
+
+ .. py:attribute:: username
+ :annotation: str
+
+ username string
+
+ .. py:attribute:: password
+ :annotation: str
+
+ password, which is normally a string but may also be any
+ object that has a ``__str__()`` method.
+
+ .. py:attribute:: host
+ :annotation: str
+
+ string hostname
+
+ .. py:attribute:: port
+ :annotation: int
+
+ integer port number
+
+ .. py:attribute:: database
+ :annotation: str
+
+ string database name
+
+ .. py:attribute:: query
+ :annotation: Mapping[str, Union[str, Sequence[str]]]
+
+ an immutable mapping representing the query string. contains strings
+ for keys and either strings or tuples of strings for values, e.g.::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
+ >>> url.query
+ immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
+
+ To create a mutable copy of this mapping, use the ``dict`` constructor::
+
+ mutable_query_opts = dict(url.query)
+
+ .. seealso::
+
+ :attr:`_engine.URL.normalized_query` - normalizes all values into sequences
+ for consistent processing
+
+ Methods for altering the contents of :attr:`_engine.URL.query`:
+
+ :meth:`_engine.URL.update_query_dict`
+
+ :meth:`_engine.URL.update_query_string`
+
+ :meth:`_engine.URL.update_query_pairs`
+
+ :meth:`_engine.URL.difference_update_query`
+
Pooling
=======
.. currentmodule:: sqlalchemy
-.. autoclass:: sqlalchemy.engine.interfaces.Compiled
+.. autoclass:: sqlalchemy.engine.Compiled
:members:
.. autoclass:: sqlalchemy.sql.compiler.DDLCompiler
:members:
:inherited-members:
-.. autoclass:: sqlalchemy.engine.interfaces.Dialect
+.. autoclass:: sqlalchemy.engine.Dialect
:members:
.. autoclass:: sqlalchemy.engine.default.DefaultExecutionContext
:members:
-.. autoclass:: sqlalchemy.engine.interfaces.ExecutionContext
+.. autoclass:: sqlalchemy.engine.ExecutionContext
:members:
-import copy
-
from ... import exc
from ...testing.provision import configure_follower
from ...testing.provision import create_db
@generate_driver_url.for_db("mysql", "mariadb")
-def generate_driver_url(url, driver, query):
+def generate_driver_url(url, driver, query_str):
backend = url.get_backend_name()
if backend == "mysql":
if dialect_cls._is_mariadb_from_url(url):
backend = "mariadb"
- new_url = copy.copy(url)
- new_url.query = dict(new_url.query)
- new_url.drivername = "%s+%s" % (backend, driver)
- new_url.query.update(query)
+ new_url = url.set(
+ drivername="%s+%s" % (backend, driver)
+ ).update_query_string(query_str)
try:
new_url.get_dialect()
@follower_url_from_main.for_db("oracle")
def _oracle_follower_url_from_main(url, ident):
url = sa_url.make_url(url)
- url.username = ident
- url.password = "xe"
- return url
+ return url.set(username=ident, password="xe")
@temp_table_keyword_args.for_db("oracle")
util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
if pysqlite_opts.get("uri", False):
- uri_opts = opts.copy()
+ uri_opts = dict(opts)
# here, we are actually separating the parameters that go to
# sqlite3/pysqlite vs. those that go the SQLite URI. What if
# two names conflict? again, this seems to be not the case right
from .row import LegacyRow # noqa
from .row import Row # noqa
from .row import RowMapping # noqa
+from .url import make_url # noqa
+from .url import URL # noqa
from .util import connection_memoize # noqa
from ..sql import ddl # noqa
echo = log.echo_property()
def __repr__(self):
- return "Engine(%r)" % self.url
+ return "Engine(%r)" % (self.url,)
def dispose(self):
"""Dispose of the connection pool used by this
# create url.URL object
u = _url.make_url(url)
- plugins = u._instantiate_plugins(kwargs)
-
- u.query.pop("plugin", None)
- kwargs.pop("plugins", None)
+ u, plugins, kwargs = u._instantiate_plugins(kwargs)
entrypoint = u._get_entrypoint()
dialect_cls = entrypoint.get_dialect_cls(u)
from .. import util
from ..sql.compiler import Compiled # noqa
from ..sql.compiler import TypeCompiler # noqa
+from ..util import compat
+
+if compat.TYPE_CHECKING:
+ from typing import Any
+ from .url import URL
class Dialect(object):
"""A set of hooks intended to augment the construction of an
:class:`_engine.Engine` object based on entrypoint names in a URL.
- The purpose of :class:`.CreateEnginePlugin` is to allow third-party
+ The purpose of :class:`_engine.CreateEnginePlugin` is to allow third-party
systems to apply engine, pool and dialect level event listeners without
the need for the target application to be modified; instead, the plugin
names can be added to the database URL. Target applications for
- :class:`.CreateEnginePlugin` include:
+ :class:`_engine.CreateEnginePlugin` include:
* connection and SQL performance tools, e.g. which use events to track
number of checkouts and/or time spent with statements
* connectivity plugins such as proxies
+ A rudimentary :class:`_engine.CreateEnginePlugin` that attaches a logger
+ to an :class:`_engine.Engine` object might look like::
+
+
+ import logging
+
+ from sqlalchemy.engine import CreateEnginePlugin
+ from sqlalchemy import event
+
+ class LogCursorEventsPlugin(CreateEnginePlugin):
+ def __init__(self, url, kwargs):
+ # consume the parameter "log_cursor_logging_name" from the
+ # URL query
+ logging_name = url.query.get("log_cursor_logging_name", "log_cursor")
+
+ self.log = logging.getLogger(logging_name)
+
+ def update_url(self, url):
+ "update the URL to one that no longer includes our parameters"
+ return url.difference_update_query(["log_cursor_logging_name"])
+
+ def engine_created(self, engine):
+ "attach an event listener after the new Engine is constructed"
+ event.listen(engine, "before_cursor_execute", self._log_event)
+
+
+ def _log_event(
+ self,
+ conn,
+ cursor,
+ statement,
+ parameters,
+ context,
+ executemany):
+
+ self.log.info("Plugin logged cursor event: %s", statement)
+
+
+
Plugins are registered using entry points in a similar way as that
of dialects::
entry_points={
'sqlalchemy.plugins': [
- 'myplugin = myapp.plugins:MyPlugin'
+ 'log_cursor_plugin = myapp.plugins:LogCursorEventsPlugin'
]
A plugin that uses the above names would be invoked from a database
from sqlalchemy import create_engine
engine = create_engine(
- "mysql+pymysql://scott:tiger@localhost/test?plugin=myplugin")
+ "mysql+pymysql://scott:tiger@localhost/test?"
+ "plugin=log_cursor_plugin&log_cursor_logging_name=mylogger"
+ )
- Alternatively, the :paramref:`_sa.create_engine.plugins" argument may be
- passed as a list to :func:`_sa.create_engine`::
+ The ``plugin`` URL parameter supports multiple instances, so that a URL
+ may specify multiple plugins; they are loaded in the order stated
+ in the URL::
+
+ engine = create_engine(
+ "mysql+pymysql://scott:tiger@localhost/test?"
+ "plugin=plugin_one&plugin=plugin_twp&plugin=plugin_three")
+
+ The plugin names may also be passed directly to :func:`_sa.create_engine`
+ using the :paramref:`_sa.create_engine.plugins` argument::
engine = create_engine(
"mysql+pymysql://scott:tiger@localhost/test",
.. versionadded:: 1.2.3 plugin names can also be specified
to :func:`_sa.create_engine` as a list
- The ``plugin`` argument supports multiple instances, so that a URL
- may specify multiple plugins; they are loaded in the order stated
- in the URL::
-
- engine = create_engine(
- "mysql+pymysql://scott:tiger@localhost/"
- "test?plugin=plugin_one&plugin=plugin_twp&plugin=plugin_three")
-
- A plugin can receive additional arguments from the URL string as
- well as from the keyword arguments passed to :func:`_sa.create_engine`.
- The :class:`.URL` object and the keyword dictionary are passed to the
- constructor so that these arguments can be extracted from the url's
- :attr:`.URL.query` collection as well as from the dictionary::
+ A plugin may consume plugin-specific arguments from the
+ :class:`_engine.URL` object as well as the ``kwargs`` dictionary, which is
+ the dictionary of arguments passed to the :func:`_sa.create_engine`
+ call. "Consuming" these arguments includes that they must be removed
+ when the plugin initializes, so that the arguments are not passed along
+ to the :class:`_engine.Dialect` constructor, where they will raise an
+ :class:`_exc.ArgumentError` because they are not known by the dialect.
+
+ As of version 1.4 of SQLAlchemy, arguments should continue to be consumed
+ from the ``kwargs`` dictionary directly, by removing the values with a
+ method such as ``dict.pop``. Arguments from the :class:`_engine.URL` object
+ should be consumed by implementing the
+ :meth:`_engine.CreateEnginePlugin.update_url` method, returning a new copy
+ of the :class:`_engine.URL` with plugin-specific parameters removed::
class MyPlugin(CreateEnginePlugin):
def __init__(self, url, kwargs):
- self.my_argument_one = url.query.pop('my_argument_one')
- self.my_argument_two = url.query.pop('my_argument_two')
+ self.my_argument_one = url.query['my_argument_one']
+ self.my_argument_two = url.query['my_argument_two']
self.my_argument_three = kwargs.pop('my_argument_three', None)
- Arguments like those illustrated above would be consumed from the
- following::
+ def update_url(self, url):
+ return url.difference_update_query(
+ ["my_argument_one", "my_argument_two"]
+ )
+
+ Arguments like those illustrated above would be consumed from a
+ :func:`_sa.create_engine` call such as::
from sqlalchemy import create_engine
engine = create_engine(
- "mysql+pymysql://scott:tiger@localhost/"
- "test?plugin=myplugin&my_argument_one=foo&my_argument_two=bar",
- my_argument_three='bat')
+ "mysql+pymysql://scott:tiger@localhost/test?"
+ "plugin=myplugin&my_argument_one=foo&my_argument_two=bar",
+ my_argument_three='bat'
+ )
+
+ .. versionchanged:: 1.4
+
+ The :class:`_engine.URL` object is now immutable; a
+ :class:`_engine.CreateEnginePlugin` that needs to alter the
+ :class:`_engine.URL` should implement the newly added
+ :meth:`_engine.CreateEnginePlugin.update_url` method, which
+ is invoked after the plugin is constructed.
+
+ For migration, construct the plugin in the following way, checking
+ for the existence of the :meth:`_engine.CreateEnginePlugin.update_url`
+ method to detect which version is running::
+
+ class MyPlugin(CreateEnginePlugin):
+ def __init__(self, url, kwargs):
+ if hasattr(CreateEnginePlugin, "update_url"):
+ # detect the 1.4 API
+ self.my_argument_one = url.query['my_argument_one']
+ self.my_argument_two = url.query['my_argument_two']
+ else:
+ # detect the 1.3 and earlier API - mutate the
+ # URL directly
+ self.my_argument_one = url.query.pop('my_argument_one')
+ self.my_argument_two = url.query.pop('my_argument_two')
+
+ self.my_argument_three = kwargs.pop('my_argument_three', None)
+
+ def update_url(self, url):
+ # this method is only called in the 1.4 version
+ return url.difference_update_query(
+ ["my_argument_one", "my_argument_two"]
+ )
+
+ .. seealso::
+
+ :ref:`change_5526` - overview of the :class:`_engine.URL` change which
+ also includes notes regarding :class:`_engine.CreateEnginePlugin`.
- The URL and dictionary are used for subsequent setup of the engine
- as they are, so the plugin can modify their arguments in-place.
- Arguments that are only understood by the plugin should be popped
- or otherwise removed so that they aren't interpreted as erroneous
- arguments afterwards.
When the engine creation process completes and produces the
:class:`_engine.Engine` object, it is again passed to the plugin via the
- :meth:`.CreateEnginePlugin.engine_created` hook. In this hook, additional
+ :meth:`_engine.CreateEnginePlugin.engine_created` hook. In this hook, additional
changes can be made to the engine, most typically involving setup of
events (e.g. those defined in :ref:`core_event_toplevel`).
.. versionadded:: 1.1
- """
+ """ # noqa: E501
def __init__(self, url, kwargs):
+ # type: (URL, dict[str: Any])
"""Construct a new :class:`.CreateEnginePlugin`.
The plugin object is instantiated individually for each call
passed to the :meth:`.CreateEnginePlugin.engine_created` method
corresponding to this URL.
- :param url: the :class:`.URL` object. The plugin should inspect
- what it needs here as well as remove its custom arguments from the
- :attr:`.URL.query` collection. The URL can be modified in-place
- in any other way as well.
+ :param url: the :class:`_engine.URL` object. The plugin may inspect
+ the :class:`_engine.URL` for arguments. Arguments used by the
+ plugin should be removed, by returning an updated :class:`_engine.URL`
+ from the :meth:`_engine.CreateEnginePlugin.update_url` method.
+
+ .. versionchanged:: 1.4
+
+ The :class:`_engine.URL` object is now immutable, so a
+ :class:`_engine.CreateEnginePlugin` that needs to alter the
+ :class:`_engine.URL` object should impliement the
+ :meth:`_engine.CreateEnginePlugin.update_url` method.
+
:param kwargs: The keyword arguments passed to :func:`.create_engine`.
- The plugin can read and modify this dictionary in-place, to affect
- the ultimate arguments used to create the engine. It should
- remove its custom arguments from the dictionary as well.
"""
self.url = url
+ def update_url(self, url):
+ """Update the :class:`_engine.URL`.
+
+ A new :class:`_engine.URL` should be returned. This method is
+ typically used to consume configuration arguments from the
+ :class:`_engine.URL` which must be removed, as they will not be
+ recognized by the dialect. The
+ :meth:`_engine.URL.difference_update_query` method is available
+ to remove these arguments. See the docstring at
+ :class:`_engine.CreateEnginePlugin` for an example.
+
+
+ .. versionadded:: 1.4
+
+ """
+
def handle_dialect_kwargs(self, dialect_cls, dialect_args):
"""parse and modify dialect kwargs"""
@_generative
def unique(self, strategy=None):
- # type(Optional[object]) -> Result
+ # type: (Optional[object]) -> Result
"""Apply unique filtering to the objects returned by this
:class:`_engine.Result`.
from .. import util
from ..dialects import plugins
from ..dialects import registry
-
-
-class URL(object):
+from ..util import collections_abc
+from ..util import compat
+
+
+if compat.TYPE_CHECKING:
+ from typing import Mapping
+ from typing import Optional
+ from typing import Sequence
+ from typing import Tuple
+ from typing import Union
+
+
+class URL(
+ util.namedtuple(
+ "URL",
+ [
+ "drivername",
+ "username",
+ "password",
+ "host",
+ "port",
+ "database",
+ "query",
+ ],
+ )
+):
"""
Represent the components of a URL used to connect to a database.
This object is suitable to be passed directly to a
- :func:`~sqlalchemy.create_engine` call. The fields of the URL are parsed
+ :func:`_sa.create_engine` call. The fields of the URL are parsed
from a string by the :func:`.make_url` function. The string
format of the URL is an RFC-1738-style string.
- All initialization parameters are available as public attributes.
+ To create a new :class:`_engine.URL` object, use the
+ :func:`_engine.url.make_url` function. To construct a :class:`_engine.URL`
+ programmatically, use the :meth:`_engine.URL.create` constructor.
- :param drivername: the name of the database backend.
- This name will correspond to a module in sqlalchemy/databases
- or a third party plug-in.
+ .. versionchanged:: 1.4
- :param username: The user name.
+ The :class:`_engine.URL` object is now an immutable object. To
+ create a URL, use the :func:`_engine.make_url` or
+ :meth:`_engine.URL.create` function / method. To modify
+ a :class:`_engine.URL`, use methods like
+ :meth:`_engine.URL.set` and
+ :meth:`_engine.URL.update_query_dict` to return a new
+ :class:`_engine.URL` object with modifications. See notes for this
+ change at :ref:`change_5526`.
- :param password: database password.
+ :class:`_engine.URL` contains the following attributes:
- :param host: The name of the host.
+ :var `_engine.URL.driver`: database backend and driver name, such as
+ ``postgresql+psycopg2``
+ :var `_engine.URL.username`: username string
+ :var `_engine.URL.password`: password, which is normally a string but may
+ also be any object that has a ``__str__()`` method.
+ :var `_engine.URL.host`: string hostname
+ :var `_engine.URL.port`: integer port number
+ :var `_engine.URL.database`: string database name
+ :var `_engine.URL.query`: an immutable mapping representing the query
+ string. contains strings for keys and either strings or tuples of strings
+ for values.
- :param port: The port number.
- :param database: The database name.
+ """
- :param query: A dictionary of options to be passed to the
- dialect and/or the DBAPI upon connect.
+ def __new__(self, *arg, **kw):
+ if not kw and len(arg) == 7:
+ return super(URL, self).__new__(self, *arg, **kw)
+ else:
+ util.warn_deprecated(
+ "Calling URL() directly is deprecated and will be disabled "
+ "in a future release. The public constructor for URL is "
+ "now the URL.create() method.",
+ "1.4",
+ )
+ return URL.create(*arg, **kw)
+
+ @classmethod
+ def create(
+ cls,
+ drivername, # type: str
+ username=None, # type: Optional[str]
+ password=None, # type: Optional[Union[str, object]]
+ host=None, # type: Optional[str]
+ port=None, # type: Optional[int]
+ database=None, # type: Optional[str]
+ query=util.EMPTY_DICT, # type: Mapping[str, Union[str, Sequence[str]]]
+ ):
+ # type: (...) -> URL
+ """Create a new :class:`_engine.URL` object.
+
+ :param drivername: the name of the database backend. This name will
+ correspond to a module in sqlalchemy/databases or a third party
+ plug-in.
+ :param username: The user name.
+ :param password: database password. May be a string or an object that
+ can be stringified with ``str()``.
+ :param host: The name of the host.
+ :param port: The port number.
+ :param database: The database name.
+ :param query: A dictionary of string keys to string values to be passed
+ to the dialect and/or the DBAPI upon connect. To specify non-string
+ parameters to a Python DBAPI directly, use the
+ :paramref:`_sa.create_engine.connect_args` parameter to
+ :func:`_sa.create_engine`. See also
+ :attr:`_engine.URL.normalized_query` for a dictionary that is
+ consistently string->list of string.
+ :return: new :class:`_engine.URL` object.
+
+ .. versionadded:: 1.4
+
+ The :class:`_engine.URL` object is now an **immutable named
+ tuple**. In addition, the ``query`` dictionary is also immutable.
+ To create a URL, use the :func:`_engine.url.make_url` or
+ :meth:`_engine.URL.create` function/ method. To modify a
+ :class:`_engine.URL`, use the :meth:`_engine.URL.set` and
+ :meth:`_engine.URL.update_query` methods.
"""
- def __init__(
+ return cls(
+ cls._assert_str(drivername, "drivername"),
+ cls._assert_none_str(username, "username"),
+ password,
+ cls._assert_none_str(host, "host"),
+ cls._assert_port(port),
+ cls._assert_none_str(database, "database"),
+ cls._str_dict(query),
+ )
+
+ @classmethod
+ def _assert_port(cls, port):
+ if port is None:
+ return None
+ try:
+ return int(port)
+ except TypeError:
+ raise TypeError("Port argument must be an integer or None")
+
+ @classmethod
+ def _assert_str(cls, v, paramname):
+ if v is None:
+ return v
+
+ if not isinstance(v, compat.string_types):
+ raise TypeError("%s must be a string" % paramname)
+ return v
+
+ @classmethod
+ def _assert_none_str(cls, v, paramname):
+ if v is None:
+ return v
+
+ return cls._assert_str(v, paramname)
+
+ @classmethod
+ def _str_dict(cls, dict_):
+ if dict_ is None:
+ return util.EMPTY_DICT
+
+ def _assert_value(val):
+ if isinstance(val, str):
+ return val
+ elif isinstance(val, collections_abc.Sequence):
+ return tuple(_assert_value(elem) for elem in val)
+ else:
+ raise TypeError(
+ "Query dictionary values must be strings or "
+ "sequences of strings"
+ )
+
+ def _assert_str(v):
+ if not isinstance(v, compat.string_types):
+ raise TypeError("Query dictionary keys must be strings")
+ return v
+
+ if isinstance(dict_, collections_abc.Sequence):
+ dict_items = dict_
+ else:
+ dict_items = dict_.items()
+
+ return util.immutabledict(
+ {
+ _assert_str(key): _assert_value(value,)
+ for key, value in dict_items
+ }
+ )
+
+ def set(
self,
- drivername,
- username=None,
- password=None,
- host=None,
- port=None,
- database=None,
- query=None,
+ drivername=None, # type: Optional[str]
+ username=None, # type: Optional[str]
+ password=None, # type: Optional[Union[str, object]]
+ host=None, # type: Optional[str]
+ port=None, # type: Optional[int]
+ database=None, # type: Optional[str]
+ query=None, # type: Optional[Mapping[str, Union[str, Sequence[str]]]]
):
- self.drivername = drivername
- self.username = username
- self.password_original = password
- self.host = host
+ # type: (...) -> URL
+ """return a new :class:`_engine.URL` object with modifications.
+
+ Values are used if they are non-None. To set a value to ``None``
+ explicitly, use the :meth:`_engine.URL._replace` method adapted
+ from ``namedtuple``.
+
+ :param drivername: new drivername
+ :param username: new username
+ :param password: new password
+ :param host: new hostname
+ :param port: new port
+ :param query: new query parameters, passed a dict of string keys
+ referring to string or sequence of string values. Fully
+ replaces the previous list of arguments.
+
+ :return: new :class:`_engine.URL` object.
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :meth:`_engine.URL.update_query_dict`
+
+ """
+
+ kw = {}
+ if drivername is not None:
+ kw["drivername"] = drivername
+ if username is not None:
+ kw["username"] = username
+ if password is not None:
+ kw["password"] = password
+ if host is not None:
+ kw["host"] = host
if port is not None:
- self.port = int(port)
+ kw["port"] = port
+ if database is not None:
+ kw["database"] = database
+ if query is not None:
+ kw["query"] = query
+
+ return self._replace(**kw)
+
+ def _replace(self, **kw):
+ # type: (**object) -> URL
+ """Override ``namedtuple._replace()`` to provide argument checking."""
+
+ if "drivername" in kw:
+ self._assert_str(kw["drivername"], "drivername")
+ for name in "username", "host", "database":
+ if name in kw:
+ self._assert_none_str(kw[name], name)
+ if "port" in kw:
+ self._assert_port(kw["port"])
+ if "query" in kw:
+ kw["query"] = self._str_dict(kw["query"])
+
+ return super(URL, self)._replace(**kw)
+
+ def update_query_string(self, query_string, append=False):
+ # type: (str, bool) -> URL
+ """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
+ parameter dictionary updated by the given query string.
+
+ E.g.::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname")
+ >>> url = url.update_query_string("alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
+ >>> str(url)
+ 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
+
+ :param query_string: a URL escaped query string, not including the
+ question mark.
+
+ :param append: if True, parameters in the existing query string will
+ not be removed; new parameters will be in addition to those present.
+ If left at its default of False, keys present in the given query
+ parameters will replace those of the existing query string.
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :attr:`_engine.URL.query`
+
+ :meth:`_engine.URL.update_query_dict`
+
+ """ # noqa: E501
+ return self.update_query_pairs(
+ util.parse_qsl(query_string), append=append
+ )
+
+ def update_query_pairs(self, key_value_pairs, append=False):
+ # type: (Sequence[Tuple[str, str]], bool) -> URL
+ """Return a new :class:`_engine.URL` object with the
+ :attr:`_engine.URL.query`
+ parameter dictionary updated by the given sequence of key/value pairs
+
+ E.g.::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname")
+ >>> url = url.update_query_pairs([("alt_host", "host1"), ("alt_host", "host2"), ("ssl_cipher", "/path/to/crt")])
+ >>> str(url)
+ 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
+
+ :param key_value_pairs: A sequence of tuples containing two strings
+ each.
+
+ :param append: if True, parameters in the existing query string will
+ not be removed; new parameters will be in addition to those present.
+ If left at its default of False, keys present in the given query
+ parameters will replace those of the existing query string.
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :attr:`_engine.URL.query`
+
+ :meth:`_engine.URL.difference_update_query`
+
+ :meth:`_engine.URL.set`
+
+ """ # noqa: E501
+
+ existing_query = self.query
+ new_keys = {}
+
+ for key, value in key_value_pairs:
+ if key in new_keys:
+ new_keys[key] = util.to_list(new_keys[key])
+ new_keys[key].append(value)
+ else:
+ new_keys[key] = value
+
+ if append:
+ new_query = {}
+
+ for k in new_keys:
+ if k in existing_query:
+ new_query[k] = util.to_list(
+ existing_query[k]
+ ) + util.to_list(new_keys[k])
+ else:
+ new_query[k] = new_keys[k]
+
+ new_query.update(
+ {
+ k: existing_query[k]
+ for k in set(existing_query).difference(new_keys)
+ }
+ )
else:
- self.port = None
- self.database = database
- self.query = query or {}
+ new_query = self.query.union(new_keys)
+ return self.set(query=new_query)
+
+ def update_query_dict(self, query_parameters, append=False):
+ # type: (Mapping[str, Union[str, Sequence[str]]], bool) -> URL
+ """Return a new :class:`_engine.URL` object with the
+ :attr:`_engine.URL.query` parameter dictionary updated by the given
+ dictionary.
+
+ The dictionary typically contains string keys and string values.
+ In order to represent a query parameter that is expressed multiple
+ times, pass a sequence of string values.
+
+ E.g.::
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname")
+ >>> url = url.update_query_dict({"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"})
+ >>> str(url)
+ 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
+
+
+ :param query_parameters: A dictionary with string keys and values
+ that are either strings, or sequences of strings.
+
+ :param append: if True, parameters in the existing query string will
+ not be removed; new parameters will be in addition to those present.
+ If left at its default of False, keys present in the given query
+ parameters will replace those of the existing query string.
+
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :attr:`_engine.URL.query`
+
+ :meth:`_engine.URL.update_query_string`
+
+ :meth:`_engine.URL.update_query_pairs`
+
+ :meth:`_engine.URL.difference_update_query`
+
+ :meth:`_engine.URL.set`
+
+ """ # noqa: E501
+ return self.update_query_pairs(query_parameters.items(), append=append)
+
+ def difference_update_query(self, names):
+ # type: (Sequence[str]) -> URL
+ """
+ Remove the given names from the :attr:`_engine.URL.query` dictionary,
+ returning the new :class:`_engine.URL`.
+
+ E.g.::
+
+ url = url.difference_update_query(['foo', 'bar'])
+
+ Equivalent to using :meth:`_engine.URL.set` as follows::
+
+ url = url.set(
+ query={
+ key: url.query[key]
+ for key in set(url.query).difference(['foo', 'bar'])
+ }
+ )
+
+ .. versionadded:: 1.4
+
+ .. seealso::
+
+ :attr:`_engine.URL.query`
+
+ :meth:`_engine.URL.update_query_dict`
+
+ :meth:`_engine.URL.set`
+
+ """
+
+ if not set(names).intersection(self.query):
+ return self
+
+ return URL(
+ self.drivername,
+ self.username,
+ self.password,
+ self.host,
+ self.port,
+ self.database,
+ util.immutabledict(
+ {
+ key: self.query[key]
+ for key in set(self.query).difference(names)
+ }
+ ),
+ )
+
+ @util.memoized_property
+ def normalized_query(self):
+ """Return the :attr:`_engine.URL.query` dictionary with values normalized
+ into sequences.
+
+ As the :attr:`_engine.URL.query` dictionary may contain either
+ string values or sequences of string values to differentiate between
+ parameters that are specified multiple times in the query string,
+ code that needs to handle multiple parameters generically will wish
+ to use this attribute so that all parameters present are presented
+ as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
+ function. E.g.::
+
+
+ >>> from sqlalchemy.engine import make_url
+ >>> url = make_url("postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
+ >>> url.query
+ immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
+ >>> url.normalized_query
+ immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
+
+ """ # noqa: E501
+
+ return util.immutabledict(
+ {
+ k: (v,) if not isinstance(v, tuple) else v
+ for k, v in self.query.items()
+ }
+ )
+
+ @util.deprecated(
+ "1.4",
+ "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
+ "be removed in a future release. Please use the "
+ ":meth:`_engine.URL.render_as_string` method.",
+ )
def __to_string__(self, hide_password=True):
+ # type: (bool) -> str
+ """Render this :class:`_engine.URL` object as a string.
+
+ :param hide_password: Defaults to True. The password is not shown
+ in the string unless this is set to False.
+
+ """
+ return self.render_as_string(hide_password=hide_password)
+
+ def render_as_string(self, hide_password=True):
+ # type: (bool) -> str
+ """Render this :class:`_engine.URL` object as a string.
+
+ This method is used when the ``__str__()`` or ``__repr__()``
+ methods are used. The method directly includes additional options.
+
+ :param hide_password: Defaults to True. The password is not shown
+ in the string unless this is set to False.
+
+ """
s = self.drivername + "://"
if self.username is not None:
s += _rfc_1738_quote(self.username)
if self.password is not None:
s += ":" + (
- "***" if hide_password else _rfc_1738_quote(self.password)
+ "***"
+ if hide_password
+ else _rfc_1738_quote(str(self.password))
)
s += "@"
if self.host is not None:
return s
def __str__(self):
- return self.__to_string__(hide_password=False)
+ return self.render_as_string(hide_password=False)
def __repr__(self):
- return self.__to_string__()
+ return self.render_as_string()
def __hash__(self):
return hash(str(self))
def __ne__(self, other):
return not self == other
- @property
- def password(self):
- if self.password_original is None:
- return None
- else:
- return util.text_type(self.password_original)
+ def get_backend_name(self):
+ """Return the backend name.
- @password.setter
- def password(self, password):
- self.password_original = password
+ This is the name that corresponds to the database backend in
+ use, and is the portion of the :attr:`_engine.URL.drivername`
+ that is to the left of the plus sign.
- def get_backend_name(self):
+ """
if "+" not in self.drivername:
return self.drivername
else:
return self.drivername.split("+")[0]
def get_driver_name(self):
+ """Return the backend name.
+
+ This is the name that corresponds to the DBAPI driver in
+ use, and is the portion of the :attr:`_engine.URL.drivername`
+ that is to the right of the plus sign.
+
+ If the :attr:`_engine.URL.drivername` does not include a plus sign,
+ then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
+ is imported in order to get the driver name.
+
+ """
+
if "+" not in self.drivername:
return self.get_dialect().driver
else:
plugin_names = util.to_list(self.query.get("plugin", ()))
plugin_names += kwargs.get("plugins", [])
- return [
+ kwargs = dict(kwargs)
+
+ loaded_plugins = [
plugins.load(plugin_name)(self, kwargs)
for plugin_name in plugin_names
]
+ u = self.difference_update_query(["plugin", "plugins"])
+
+ for plugin in loaded_plugins:
+ new_u = plugin.update_url(u)
+ if new_u is not None:
+ u = new_u
+
+ kwargs.pop("plugins", None)
+
+ return u, loaded_plugins, kwargs
+
def _get_entrypoint(self):
"""Return the "entry point" dialect class.
return cls
def get_dialect(self):
- """Return the SQLAlchemy database dialect class corresponding
+ """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
to this URL's driver name.
+
"""
entrypoint = self._get_entrypoint()
dialect_cls = entrypoint.get_dialect_cls(self)
ipv6host = components.pop("ipv6host")
components["host"] = ipv4host or ipv6host
name = components.pop("name")
- return URL(name, **components)
+
+ if components["port"]:
+ components["port"] = int(components["port"])
+
+ return URL.create(name, **components)
+
else:
raise exc.ArgumentError(
"Could not parse rfc1738 URL from string '%s'" % name
elif dialect == "default_enhanced":
dialect = default.StrCompileDialect()
elif isinstance(dialect, util.string_types):
- dialect = url.URL(dialect).get_dialect()()
+ dialect = url.URL.create(dialect).get_dialect()()
if default_schema_name:
dialect.default_schema_name = default_schema_name
params = {"implicit_returning": True}
else:
params = {}
- return url.URL(self.dialect).get_dialect()(**params)
+ return url.URL.create(self.dialect).get_dialect()(**params)
def _received_statement(self, execute_observed):
"""reconstruct the statement and params in terms
import collections
-import copy
import logging
from . import config
from .. import exc
from ..engine import url as sa_url
from ..util import compat
-from ..util import parse_qsl
log = logging.getLogger(__name__)
FOLLOWER_IDENT = None
+if compat.TYPE_CHECKING:
+ from ..engine import URL
+
class register(object):
def __init__(self):
main_driver = url.get_driver_name()
extra_drivers.discard(main_driver)
- url = generate_driver_url(url, main_driver, {})
+ url = generate_driver_url(url, main_driver, "")
yield str(url)
for drv in list(extra_drivers):
driver_only, query_str = drv.split("?", 1)
- query = parse_qsl(query_str)
else:
driver_only = drv
- query = {}
+ query_str = None
- new_url = generate_driver_url(url, driver_only, query)
+ new_url = generate_driver_url(url, driver_only, query_str)
if new_url:
extra_drivers.remove(drv)
@register.init
-def generate_driver_url(url, driver, query):
+def generate_driver_url(url, driver, query_str):
+ # type: (URL, str, str) -> URL
backend = url.get_backend_name()
- new_url = copy.copy(url)
- new_url.query = dict(new_url.query)
- new_url.drivername = "%s+%s" % (backend, driver)
- new_url.query.update(query)
+
+ new_url = url.set(drivername="%s+%s" % (backend, driver),)
+ new_url = new_url.update_query_string(query_str)
+
try:
new_url.get_dialect()
except exc.NoSuchModuleError:
database name
"""
url = sa_url.make_url(url)
- url.database = ident
- return url
+ return url.set(database=ident)
@register.init
port=None,
query={},
database=None,
- _instantiate_plugins=lambda kw: [],
+ _instantiate_plugins=lambda kw: (u1, [], kw),
_get_entrypoint=Mock(
return_value=Mock(get_dialect_cls=lambda u: SomeDialect)
),
from sqlalchemy.engine.default import DefaultDialect
import sqlalchemy.engine.url as url
from sqlalchemy.testing import assert_raises
+from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
return self.value
sp = SecurePassword("secured_password")
- u = url.URL("dbtype", username="x", password=sp, host="localhost")
+ u = url.URL.create(
+ "dbtype", username="x", password=sp, host="localhost"
+ )
eq_(u.password, "secured_password")
eq_(str(u), "dbtype://x:secured_password@localhost")
# test in-place modification
sp.value = "new_secured_password"
- eq_(u.password, "new_secured_password")
+ eq_(u.password, sp)
eq_(str(u), "dbtype://x:new_secured_password@localhost")
- u.password = "hi"
+ u = u.set(password="hi")
eq_(u.password, "hi")
eq_(str(u), "dbtype://x:hi@localhost")
- u.password = None
+ u = u._replace(password=None)
is_(u.password, None)
eq_(str(u), "dbtype://x@localhost")
u = url.make_url(
"dialect://user:pass@host/db?arg1=param1&arg2=param2&arg2=param3"
)
- eq_(u.query, {"arg1": "param1", "arg2": ["param2", "param3"]})
+ eq_(u.query, {"arg1": "param1", "arg2": ("param2", "param3")})
eq_(
str(u),
"dialect://user:pass@host/db?arg1=param1&arg2=param2&arg2=param3",
eq_(str(u), test_url)
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"
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))
+ @testing.combinations(
+ "drivername", "username", "password", "host", "database",
+ )
+ def test_component_set(self, component):
+ common_url = (
+ "dbtype://username:password"
+ "@[2001:da8:2004:1000:202:116:160:90]:80/database?foo=bar"
+ )
+ url1 = url.make_url(common_url)
+ url2 = url.make_url(common_url)
+
+ url3 = url2.set(**{component: "new_changed_value"})
+ is_true(url1 != url3)
+ is_false(url1 == url3)
+
+ url4 = url3.set(**{component: getattr(url1, component)})
+
+ is_true(url4 == url1)
+ is_false(url4 != url1)
+
+ @testing.combinations(
+ (
+ "foo1=bar1&foo2=bar2",
+ {"foo2": "bar22", "foo3": "bar3"},
+ "foo1=bar1&foo2=bar22&foo3=bar3",
+ False,
+ ),
+ (
+ "foo1=bar1&foo2=bar2",
+ {"foo2": "bar22", "foo3": "bar3"},
+ "foo1=bar1&foo2=bar2&foo2=bar22&foo3=bar3",
+ True,
+ ),
+ )
+ def test_update_query_dict(self, starting, update_with, expected, append):
+ eq_(
+ url.make_url("drivername:///?%s" % starting).update_query_dict(
+ update_with, append=append
+ ),
+ url.make_url("drivername:///?%s" % expected),
+ )
+
+ @testing.combinations(
+ (
+ "foo1=bar1&foo2=bar2",
+ "foo2=bar22&foo3=bar3",
+ "foo1=bar1&foo2=bar22&foo3=bar3",
+ False,
+ ),
+ (
+ "foo1=bar1&foo2=bar2",
+ "foo2=bar22&foo3=bar3",
+ "foo1=bar1&foo2=bar2&foo2=bar22&foo3=bar3",
+ True,
+ ),
+ (
+ "foo1=bar1&foo2=bar21&foo2=bar22&foo3=bar31",
+ "foo2=bar23&foo3=bar32&foo3=bar33",
+ "foo1=bar1&foo2=bar21&foo2=bar22&foo2=bar23&"
+ "foo3=bar31&foo3=bar32&foo3=bar33",
+ True,
+ ),
+ (
+ "foo1=bar1&foo2=bar21&foo2=bar22&foo3=bar31",
+ "foo2=bar23&foo3=bar32&foo3=bar33",
+ "foo1=bar1&foo2=bar23&" "foo3=bar32&foo3=bar33",
+ False,
+ ),
+ )
+ def test_update_query_string(
+ self, starting, update_with, expected, append
+ ):
+ eq_(
+ url.make_url("drivername:///?%s" % starting).update_query_string(
+ update_with, append=append
+ ),
+ url.make_url("drivername:///?%s" % expected),
+ )
+
+ @testing.combinations(
+ "username", "host", "database",
+ )
+ def test_only_str_constructor(self, argname):
+ assert_raises_message(
+ TypeError,
+ "%s must be a string" % argname,
+ url.URL.create,
+ "somedriver",
+ **{argname: 35.8}
+ )
+
+ @testing.combinations(
+ "username", "host", "database",
+ )
+ def test_only_str_set(self, argname):
+ u1 = url.URL.create("somedriver")
+
+ assert_raises_message(
+ TypeError,
+ "%s must be a string" % argname,
+ u1.set,
+ **{argname: 35.8}
+ )
+
+ def test_only_str_query_key_constructor(self):
+ assert_raises_message(
+ TypeError,
+ "Query dictionary keys must be strings",
+ url.URL.create,
+ "somedriver",
+ query={35.8: "foo"},
+ )
+
+ def test_only_str_query_value_constructor(self):
+ assert_raises_message(
+ TypeError,
+ "Query dictionary values must be strings or sequences of strings",
+ url.URL.create,
+ "somedriver",
+ query={"foo": 35.8},
+ )
+
+ def test_only_str_query_key_update(self):
+ assert_raises_message(
+ TypeError,
+ "Query dictionary keys must be strings",
+ url.make_url("drivername://").update_query_dict,
+ {35.8: "foo"},
+ )
+
+ def test_only_str_query_value_update(self):
+ assert_raises_message(
+ TypeError,
+ "Query dictionary values must be strings or sequences of strings",
+ url.make_url("drivername://").update_query_dict,
+ {"foo": 35.8},
+ )
+
+ def test_deprecated_constructor(self):
+ with testing.expect_deprecated(
+ r"Calling URL\(\) directly is deprecated and will be "
+ "disabled in a future release."
+ ):
+ u1 = url.URL(
+ drivername="somedriver",
+ username="user",
+ port=52,
+ host="hostname",
+ )
+ eq_(u1, url.make_url("somedriver://user@hostname:52"))
class DialectImportTest(fixtures.TestBase):
@testing.requires.sqlite
def test_wrapper_hooks(self):
def get_dialect_cls(url):
- url.drivername = "sqlite"
+ url = url.set(drivername="sqlite")
return url.get_dialect()
global WrapperFactory
eq_(
WrapperFactory.mock_calls,
[
- call.get_dialect_cls(url.make_url("sqlite://")),
+ call.get_dialect_cls(url.make_url("wrapperdialect://")),
call.engine_created(e),
],
)
)
eq_(kw, {"logging_name": "foob"})
kw["logging_name"] = "bar"
- url.query.pop("myplugin_arg", None)
return MyEnginePlugin
- MyEnginePlugin = Mock(side_effect=side_effect)
+ def update_url(url):
+ return url.difference_update_query(["myplugin_arg"])
+
+ MyEnginePlugin = Mock(side_effect=side_effect, update_url=update_url)
plugins.register("engineplugin", __name__, "MyEnginePlugin")
eq_(
MyEnginePlugin.mock_calls,
[
- call(url.make_url("sqlite:///?foo=bar"), {}),
+ call(
+ url.make_url(
+ "sqlite:///?plugin=engineplugin"
+ "&foo=bar&myplugin_arg=bat"
+ ),
+ {},
+ ),
call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
call.engine_created(e),
],
)
- # url was modified in place by MyEnginePlugin
- eq_(str(MyEnginePlugin.mock_calls[0][1][0]), "sqlite:///?foo=bar")
-
@testing.requires.sqlite
def test_plugin_multiple_url_registration(self):
from sqlalchemy.dialects import sqlite
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)
+ def update_url(url):
+ return url.difference_update_query(
+ ["myplugin1_arg", "myplugin2_arg"]
+ )
+
+ MyEnginePlugin1 = Mock(
+ side_effect=side_effect_1, update_url=update_url
+ )
+ MyEnginePlugin2 = Mock(
+ side_effect=side_effect_2, update_url=update_url
+ )
plugins.register("engineplugin1", __name__, "MyEnginePlugin1")
plugins.register("engineplugin2", __name__, "MyEnginePlugin2")
- e = create_engine(
+ url_str = (
"sqlite:///?plugin=engineplugin1&foo=bar&myplugin1_arg=bat"
- "&plugin=engineplugin2&myplugin2_arg=hoho",
- logging_name="foob",
+ "&plugin=engineplugin2&myplugin2_arg=hoho"
)
+ e = create_engine(url_str, logging_name="foob",)
eq_(e.dialect.name, "sqlite")
eq_(e.logging_name, "bar")
eq_(
MyEnginePlugin1.mock_calls,
[
- call(url.make_url("sqlite:///?foo=bar"), {}),
+ call(url.make_url(url_str), {}),
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(url.make_url(url_str), {}),
call.handle_dialect_kwargs(sqlite.dialect, mock.ANY),
call.handle_pool_kwargs(mock.ANY, {"dialect": e.dialect}),
call.engine_created(e),
kw.pop("myplugin_arg", None)
return MyEnginePlugin
- MyEnginePlugin = Mock(side_effect=side_effect)
+ def update_url(url):
+ return url.difference_update_query(["myplugin_arg"])
+
+ MyEnginePlugin = Mock(side_effect=side_effect, update_url=update_url)
plugins.register("engineplugin", __name__, "MyEnginePlugin")
class Dialect(DefaultDialect):
initialize = Mock()
- engine = create_engine(MyURL("foo://"), module=dbapi)
+ engine = create_engine(MyURL.create("foo://"), module=dbapi)
engine.connect()
# note that the dispose() call replaces the old pool with a new one;
# on a subsequent attempt without initialization having proceeded.
Dialect.initialize.side_effect = TypeError
- engine = create_engine(MyURL("foo://"), module=dbapi)
+ engine = create_engine(MyURL.create("foo://"), module=dbapi)
assert_raises(TypeError, engine.connect)
eq_(Dialect.initialize.call_count, 1)
url = Mock(
get_dialect=lambda: default.DefaultDialect,
_get_entrypoint=lambda: default.DefaultDialect,
- _instantiate_plugins=lambda kwargs: (),
+ _instantiate_plugins=lambda kwargs: (url, [], kwargs),
translate_connect_args=lambda: {},
query={},
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
