diff options
Diffstat (limited to 'lib/sqlalchemy/engine/url.py')
| -rw-r--r-- | lib/sqlalchemy/engine/url.py | 579 |
1 files changed, 532 insertions, 47 deletions
diff --git a/lib/sqlalchemy/engine/url.py b/lib/sqlalchemy/engine/url.py index f0685d9e3..6d2f4aa24 100644 --- a/lib/sqlalchemy/engine/url.py +++ b/lib/sqlalchemy/engine/url.py @@ -21,66 +21,524 @@ from .. import exc 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: @@ -103,10 +561,10 @@ class URL(object): 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)) @@ -126,24 +584,32 @@ class URL(object): 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: @@ -153,11 +619,24 @@ class URL(object): 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. @@ -183,8 +662,9 @@ class URL(object): 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) @@ -285,7 +765,12 @@ def _parse_rfc1738_args(name): 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 |