Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/url.py: 39%
238 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1# engine/url.py
2# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Provides the :class:`~sqlalchemy.engine.url.URL` class which encapsulates
9information about a database connection specification.
11The URL object is created automatically when
12:func:`~sqlalchemy.engine.create_engine` is called with a string
13argument; alternatively, the URL is a public-facing construct which can
14be used directly and is also accepted directly by ``create_engine()``.
15"""
17import re
19from .interfaces import Dialect
20from .. import exc
21from .. import util
22from ..dialects import plugins
23from ..dialects import registry
24from ..util import collections_abc
25from ..util import compat
28class URL(
29 util.namedtuple(
30 "URL",
31 [
32 "drivername",
33 "username",
34 "password",
35 "host",
36 "port",
37 "database",
38 "query",
39 ],
40 )
41):
42 """
43 Represent the components of a URL used to connect to a database.
45 URLs are typically constructed from a fully formatted URL string, where the
46 :func:`.make_url` function is used internally by the
47 :func:`_sa.create_engine` function in order to parse the URL string into
48 its individual components, which are then used to construct a new
49 :class:`.URL` object. When parsing from a formatted URL string, the parsing
50 format generally follows
51 `RFC-1738 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions.
53 A :class:`_engine.URL` object may also be produced directly, either by
54 using the :func:`.make_url` function with a fully formed URL string, or
55 by using the :meth:`_engine.URL.create` constructor in order
56 to construct a :class:`_engine.URL` programmatically given individual
57 fields. The resulting :class:`.URL` object may be passed directly to
58 :func:`_sa.create_engine` in place of a string argument, which will bypass
59 the usage of :func:`.make_url` within the engine's creation process.
61 .. versionchanged:: 1.4
63 The :class:`_engine.URL` object is now an immutable object. To
64 create a URL, use the :func:`_engine.make_url` or
65 :meth:`_engine.URL.create` function / method. To modify
66 a :class:`_engine.URL`, use methods like
67 :meth:`_engine.URL.set` and
68 :meth:`_engine.URL.update_query_dict` to return a new
69 :class:`_engine.URL` object with modifications. See notes for this
70 change at :ref:`change_5526`.
72 :class:`_engine.URL` contains the following attributes:
74 * :attr:`_engine.URL.drivername`: database backend and driver name, such as
75 ``postgresql+psycopg2``
76 * :attr:`_engine.URL.username`: username string
77 * :attr:`_engine.URL.password`: password string
78 * :attr:`_engine.URL.host`: string hostname
79 * :attr:`_engine.URL.port`: integer port number
80 * :attr:`_engine.URL.database`: string database name
81 * :attr:`_engine.URL.query`: an immutable mapping representing the query
82 string. contains strings for keys and either strings or tuples of
83 strings for values.
86 """
88 def __new__(self, *arg, **kw):
89 if kw.pop("_new_ok", False):
90 return super(URL, self).__new__(self, *arg, **kw)
91 else:
92 util.warn_deprecated(
93 "Calling URL() directly is deprecated and will be disabled "
94 "in a future release. The public constructor for URL is "
95 "now the URL.create() method.",
96 "1.4",
97 )
98 return URL.create(*arg, **kw)
100 @classmethod
101 def create(
102 cls,
103 drivername,
104 username=None,
105 password=None,
106 host=None,
107 port=None,
108 database=None,
109 query=util.EMPTY_DICT,
110 ):
111 """Create a new :class:`_engine.URL` object.
113 :param drivername: the name of the database backend. This name will
114 correspond to a module in sqlalchemy/databases or a third party
115 plug-in.
116 :param username: The user name.
117 :param password: database password. Is typically a string, but may
118 also be an object that can be stringified with ``str()``.
120 .. note:: A password-producing object will be stringified only
121 **once** per :class:`_engine.Engine` object. For dynamic password
122 generation per connect, see :ref:`engines_dynamic_tokens`.
124 :param host: The name of the host.
125 :param port: The port number.
126 :param database: The database name.
127 :param query: A dictionary of string keys to string values to be passed
128 to the dialect and/or the DBAPI upon connect. To specify non-string
129 parameters to a Python DBAPI directly, use the
130 :paramref:`_sa.create_engine.connect_args` parameter to
131 :func:`_sa.create_engine`. See also
132 :attr:`_engine.URL.normalized_query` for a dictionary that is
133 consistently string->list of string.
134 :return: new :class:`_engine.URL` object.
136 .. versionadded:: 1.4
138 The :class:`_engine.URL` object is now an **immutable named
139 tuple**. In addition, the ``query`` dictionary is also immutable.
140 To create a URL, use the :func:`_engine.url.make_url` or
141 :meth:`_engine.URL.create` function/ method. To modify a
142 :class:`_engine.URL`, use the :meth:`_engine.URL.set` and
143 :meth:`_engine.URL.update_query` methods.
145 """
147 return cls(
148 cls._assert_str(drivername, "drivername"),
149 cls._assert_none_str(username, "username"),
150 password,
151 cls._assert_none_str(host, "host"),
152 cls._assert_port(port),
153 cls._assert_none_str(database, "database"),
154 cls._str_dict(query),
155 _new_ok=True,
156 )
158 @classmethod
159 def _assert_port(cls, port):
160 if port is None:
161 return None
162 try:
163 return int(port)
164 except TypeError:
165 raise TypeError("Port argument must be an integer or None")
167 @classmethod
168 def _assert_str(cls, v, paramname):
169 if not isinstance(v, compat.string_types):
170 raise TypeError("%s must be a string" % paramname)
171 return v
173 @classmethod
174 def _assert_none_str(cls, v, paramname):
175 if v is None:
176 return v
178 return cls._assert_str(v, paramname)
180 @classmethod
181 def _str_dict(cls, dict_):
182 if dict_ is None:
183 return util.EMPTY_DICT
185 def _assert_value(val):
186 if isinstance(val, compat.string_types):
187 return val
188 elif isinstance(val, collections_abc.Sequence):
189 return tuple(_assert_value(elem) for elem in val)
190 else:
191 raise TypeError(
192 "Query dictionary values must be strings or "
193 "sequences of strings"
194 )
196 def _assert_str(v):
197 if not isinstance(v, compat.string_types):
198 raise TypeError("Query dictionary keys must be strings")
199 return v
201 if isinstance(dict_, collections_abc.Sequence):
202 dict_items = dict_
203 else:
204 dict_items = dict_.items()
206 return util.immutabledict(
207 {
208 _assert_str(key): _assert_value(
209 value,
210 )
211 for key, value in dict_items
212 }
213 )
215 def set(
216 self,
217 drivername=None,
218 username=None,
219 password=None,
220 host=None,
221 port=None,
222 database=None,
223 query=None,
224 ):
225 """return a new :class:`_engine.URL` object with modifications.
227 Values are used if they are non-None. To set a value to ``None``
228 explicitly, use the :meth:`_engine.URL._replace` method adapted
229 from ``namedtuple``.
231 :param drivername: new drivername
232 :param username: new username
233 :param password: new password
234 :param host: new hostname
235 :param port: new port
236 :param query: new query parameters, passed a dict of string keys
237 referring to string or sequence of string values. Fully
238 replaces the previous list of arguments.
240 :return: new :class:`_engine.URL` object.
242 .. versionadded:: 1.4
244 .. seealso::
246 :meth:`_engine.URL.update_query_dict`
248 """
250 kw = {}
251 if drivername is not None:
252 kw["drivername"] = drivername
253 if username is not None:
254 kw["username"] = username
255 if password is not None:
256 kw["password"] = password
257 if host is not None:
258 kw["host"] = host
259 if port is not None:
260 kw["port"] = port
261 if database is not None:
262 kw["database"] = database
263 if query is not None:
264 kw["query"] = query
266 return self._replace(**kw)
268 def _replace(self, **kw):
269 """Override ``namedtuple._replace()`` to provide argument checking."""
271 if "drivername" in kw:
272 self._assert_str(kw["drivername"], "drivername")
273 for name in "username", "host", "database":
274 if name in kw:
275 self._assert_none_str(kw[name], name)
276 if "port" in kw:
277 self._assert_port(kw["port"])
278 if "query" in kw:
279 kw["query"] = self._str_dict(kw["query"])
281 return super(URL, self)._replace(**kw)
283 def update_query_string(self, query_string, append=False):
284 """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
285 parameter dictionary updated by the given query string.
287 E.g.::
289 >>> from sqlalchemy.engine import make_url
290 >>> url = make_url("postgresql://user:pass@host/dbname")
291 >>> url = url.update_query_string("alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
292 >>> str(url)
293 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
295 :param query_string: a URL escaped query string, not including the
296 question mark.
298 :param append: if True, parameters in the existing query string will
299 not be removed; new parameters will be in addition to those present.
300 If left at its default of False, keys present in the given query
301 parameters will replace those of the existing query string.
303 .. versionadded:: 1.4
305 .. seealso::
307 :attr:`_engine.URL.query`
309 :meth:`_engine.URL.update_query_dict`
311 """ # noqa: E501
312 return self.update_query_pairs(
313 util.parse_qsl(query_string), append=append
314 )
316 def update_query_pairs(self, key_value_pairs, append=False):
317 """Return a new :class:`_engine.URL` object with the
318 :attr:`_engine.URL.query`
319 parameter dictionary updated by the given sequence of key/value pairs
321 E.g.::
323 >>> from sqlalchemy.engine import make_url
324 >>> url = make_url("postgresql://user:pass@host/dbname")
325 >>> url = url.update_query_pairs([("alt_host", "host1"), ("alt_host", "host2"), ("ssl_cipher", "/path/to/crt")])
326 >>> str(url)
327 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
329 :param key_value_pairs: A sequence of tuples containing two strings
330 each.
332 :param append: if True, parameters in the existing query string will
333 not be removed; new parameters will be in addition to those present.
334 If left at its default of False, keys present in the given query
335 parameters will replace those of the existing query string.
337 .. versionadded:: 1.4
339 .. seealso::
341 :attr:`_engine.URL.query`
343 :meth:`_engine.URL.difference_update_query`
345 :meth:`_engine.URL.set`
347 """ # noqa: E501
349 existing_query = self.query
350 new_keys = {}
352 for key, value in key_value_pairs:
353 if key in new_keys:
354 new_keys[key] = util.to_list(new_keys[key])
355 new_keys[key].append(value)
356 else:
357 new_keys[key] = value
359 if append:
360 new_query = {}
362 for k in new_keys:
363 if k in existing_query:
364 new_query[k] = util.to_list(
365 existing_query[k]
366 ) + util.to_list(new_keys[k])
367 else:
368 new_query[k] = new_keys[k]
370 new_query.update(
371 {
372 k: existing_query[k]
373 for k in set(existing_query).difference(new_keys)
374 }
375 )
376 else:
377 new_query = self.query.union(new_keys)
378 return self.set(query=new_query)
380 def update_query_dict(self, query_parameters, append=False):
381 """Return a new :class:`_engine.URL` object with the
382 :attr:`_engine.URL.query` parameter dictionary updated by the given
383 dictionary.
385 The dictionary typically contains string keys and string values.
386 In order to represent a query parameter that is expressed multiple
387 times, pass a sequence of string values.
389 E.g.::
392 >>> from sqlalchemy.engine import make_url
393 >>> url = make_url("postgresql://user:pass@host/dbname")
394 >>> url = url.update_query_dict({"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"})
395 >>> str(url)
396 'postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
399 :param query_parameters: A dictionary with string keys and values
400 that are either strings, or sequences of strings.
402 :param append: if True, parameters in the existing query string will
403 not be removed; new parameters will be in addition to those present.
404 If left at its default of False, keys present in the given query
405 parameters will replace those of the existing query string.
408 .. versionadded:: 1.4
410 .. seealso::
412 :attr:`_engine.URL.query`
414 :meth:`_engine.URL.update_query_string`
416 :meth:`_engine.URL.update_query_pairs`
418 :meth:`_engine.URL.difference_update_query`
420 :meth:`_engine.URL.set`
422 """ # noqa: E501
423 return self.update_query_pairs(query_parameters.items(), append=append)
425 def difference_update_query(self, names):
426 """
427 Remove the given names from the :attr:`_engine.URL.query` dictionary,
428 returning the new :class:`_engine.URL`.
430 E.g.::
432 url = url.difference_update_query(['foo', 'bar'])
434 Equivalent to using :meth:`_engine.URL.set` as follows::
436 url = url.set(
437 query={
438 key: url.query[key]
439 for key in set(url.query).difference(['foo', 'bar'])
440 }
441 )
443 .. versionadded:: 1.4
445 .. seealso::
447 :attr:`_engine.URL.query`
449 :meth:`_engine.URL.update_query_dict`
451 :meth:`_engine.URL.set`
453 """
455 if not set(names).intersection(self.query):
456 return self
458 return URL(
459 self.drivername,
460 self.username,
461 self.password,
462 self.host,
463 self.port,
464 self.database,
465 util.immutabledict(
466 {
467 key: self.query[key]
468 for key in set(self.query).difference(names)
469 }
470 ),
471 _new_ok=True,
472 )
474 @util.memoized_property
475 def normalized_query(self):
476 """Return the :attr:`_engine.URL.query` dictionary with values normalized
477 into sequences.
479 As the :attr:`_engine.URL.query` dictionary may contain either
480 string values or sequences of string values to differentiate between
481 parameters that are specified multiple times in the query string,
482 code that needs to handle multiple parameters generically will wish
483 to use this attribute so that all parameters present are presented
484 as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
485 function. E.g.::
488 >>> from sqlalchemy.engine import make_url
489 >>> url = make_url("postgresql://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
490 >>> url.query
491 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
492 >>> url.normalized_query
493 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
495 """ # noqa: E501
497 return util.immutabledict(
498 {
499 k: (v,) if not isinstance(v, tuple) else v
500 for k, v in self.query.items()
501 }
502 )
504 @util.deprecated(
505 "1.4",
506 "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
507 "be removed in a future release. Please use the "
508 ":meth:`_engine.URL.render_as_string` method.",
509 )
510 def __to_string__(self, hide_password=True):
511 """Render this :class:`_engine.URL` object as a string.
513 :param hide_password: Defaults to True. The password is not shown
514 in the string unless this is set to False.
516 """
517 return self.render_as_string(hide_password=hide_password)
519 def render_as_string(self, hide_password=True):
520 """Render this :class:`_engine.URL` object as a string.
522 This method is used when the ``__str__()`` or ``__repr__()``
523 methods are used. The method directly includes additional options.
525 :param hide_password: Defaults to True. The password is not shown
526 in the string unless this is set to False.
528 """
529 s = self.drivername + "://"
530 if self.username is not None:
531 s += _sqla_url_quote(self.username)
532 if self.password is not None:
533 s += ":" + (
534 "***"
535 if hide_password
536 else _sqla_url_quote(str(self.password))
537 )
538 s += "@"
539 if self.host is not None:
540 if ":" in self.host:
541 s += "[%s]" % self.host
542 else:
543 s += self.host
544 if self.port is not None:
545 s += ":" + str(self.port)
546 if self.database is not None:
547 s += "/" + self.database
548 if self.query:
549 keys = list(self.query)
550 keys.sort()
551 s += "?" + "&".join(
552 "%s=%s" % (util.quote_plus(k), util.quote_plus(element))
553 for k in keys
554 for element in util.to_list(self.query[k])
555 )
556 return s
558 def __str__(self):
559 return self.render_as_string(hide_password=False)
561 def __repr__(self):
562 return self.render_as_string()
564 def __copy__(self):
565 return self.__class__.create(
566 self.drivername,
567 self.username,
568 self.password,
569 self.host,
570 self.port,
571 self.database,
572 # note this is an immutabledict of str-> str / tuple of str,
573 # also fully immutable. does not require deepcopy
574 self.query,
575 )
577 def __deepcopy__(self, memo):
578 return self.__copy__()
580 def __hash__(self):
581 return hash(str(self))
583 def __eq__(self, other):
584 return (
585 isinstance(other, URL)
586 and self.drivername == other.drivername
587 and self.username == other.username
588 and self.password == other.password
589 and self.host == other.host
590 and self.database == other.database
591 and self.query == other.query
592 and self.port == other.port
593 )
595 def __ne__(self, other):
596 return not self == other
598 def get_backend_name(self):
599 """Return the backend name.
601 This is the name that corresponds to the database backend in
602 use, and is the portion of the :attr:`_engine.URL.drivername`
603 that is to the left of the plus sign.
605 """
606 if "+" not in self.drivername:
607 return self.drivername
608 else:
609 return self.drivername.split("+")[0]
611 def get_driver_name(self):
612 """Return the backend name.
614 This is the name that corresponds to the DBAPI driver in
615 use, and is the portion of the :attr:`_engine.URL.drivername`
616 that is to the right of the plus sign.
618 If the :attr:`_engine.URL.drivername` does not include a plus sign,
619 then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
620 is imported in order to get the driver name.
622 """
624 if "+" not in self.drivername:
625 return self.get_dialect().driver
626 else:
627 return self.drivername.split("+")[1]
629 def _instantiate_plugins(self, kwargs):
630 plugin_names = util.to_list(self.query.get("plugin", ()))
631 plugin_names += kwargs.get("plugins", [])
633 kwargs = dict(kwargs)
635 loaded_plugins = [
636 plugins.load(plugin_name)(self, kwargs)
637 for plugin_name in plugin_names
638 ]
640 u = self.difference_update_query(["plugin", "plugins"])
642 for plugin in loaded_plugins:
643 new_u = plugin.update_url(u)
644 if new_u is not None:
645 u = new_u
647 kwargs.pop("plugins", None)
649 return u, loaded_plugins, kwargs
651 def _get_entrypoint(self):
652 """Return the "entry point" dialect class.
654 This is normally the dialect itself except in the case when the
655 returned class implements the get_dialect_cls() method.
657 """
658 if "+" not in self.drivername:
659 name = self.drivername
660 else:
661 name = self.drivername.replace("+", ".")
662 cls = registry.load(name)
663 # check for legacy dialects that
664 # would return a module with 'dialect' as the
665 # actual class
666 if (
667 hasattr(cls, "dialect")
668 and isinstance(cls.dialect, type)
669 and issubclass(cls.dialect, Dialect)
670 ):
671 return cls.dialect
672 else:
673 return cls
675 def get_dialect(self):
676 """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
677 to this URL's driver name.
679 """
680 entrypoint = self._get_entrypoint()
681 dialect_cls = entrypoint.get_dialect_cls(self)
682 return dialect_cls
684 def translate_connect_args(self, names=None, **kw):
685 r"""Translate url attributes into a dictionary of connection arguments.
687 Returns attributes of this url (`host`, `database`, `username`,
688 `password`, `port`) as a plain dictionary. The attribute names are
689 used as the keys by default. Unset or false attributes are omitted
690 from the final dictionary.
692 :param \**kw: Optional, alternate key names for url attributes.
694 :param names: Deprecated. Same purpose as the keyword-based alternate
695 names, but correlates the name to the original positionally.
696 """
698 if names is not None:
699 util.warn_deprecated(
700 "The `URL.translate_connect_args.name`s parameter is "
701 "deprecated. Please pass the "
702 "alternate names as kw arguments.",
703 "1.4",
704 )
706 translated = {}
707 attribute_names = ["host", "database", "username", "password", "port"]
708 for sname in attribute_names:
709 if names:
710 name = names.pop(0)
711 elif sname in kw:
712 name = kw[sname]
713 else:
714 name = sname
715 if name is not None and getattr(self, sname, False):
716 if sname == "password":
717 translated[name] = str(getattr(self, sname))
718 else:
719 translated[name] = getattr(self, sname)
721 return translated
724def make_url(name_or_url):
725 """Given a string or unicode instance, produce a new URL instance.
728 The format of the URL generally follows `RFC-1738
729 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
730 that underscores, and not dashes or periods, are accepted within the
731 "scheme" portion.
733 If a :class:`.URL` object is passed, it is returned as is.
735 """
737 if isinstance(name_or_url, util.string_types):
738 return _parse_url(name_or_url)
739 else:
740 return name_or_url
743def _parse_url(name):
744 pattern = re.compile(
745 r"""
746 (?P<name>[\w\+]+)://
747 (?:
748 (?P<username>[^:/]*)
749 (?::(?P<password>[^@]*))?
750 @)?
751 (?:
752 (?:
753 \[(?P<ipv6host>[^/\?]+)\] |
754 (?P<ipv4host>[^/:\?]+)
755 )?
756 (?::(?P<port>[^/\?]*))?
757 )?
758 (?:/(?P<database>[^\?]*))?
759 (?:\?(?P<query>.*))?
760 """,
761 re.X,
762 )
764 m = pattern.match(name)
765 if m is not None:
766 components = m.groupdict()
767 if components["query"] is not None:
768 query = {}
770 for key, value in util.parse_qsl(components["query"]):
771 if util.py2k:
772 key = key.encode("ascii")
773 if key in query:
774 query[key] = util.to_list(query[key])
775 query[key].append(value)
776 else:
777 query[key] = value
778 else:
779 query = None
780 components["query"] = query
782 if components["username"] is not None:
783 components["username"] = _sqla_url_unquote(components["username"])
785 if components["password"] is not None:
786 components["password"] = _sqla_url_unquote(components["password"])
788 ipv4host = components.pop("ipv4host")
789 ipv6host = components.pop("ipv6host")
790 components["host"] = ipv4host or ipv6host
791 name = components.pop("name")
793 if components["port"]:
794 components["port"] = int(components["port"])
796 return URL.create(name, **components)
798 else:
799 raise exc.ArgumentError(
800 "Could not parse SQLAlchemy URL from string '%s'" % name
801 )
804def _sqla_url_quote(text):
805 return re.sub(r"[:@/]", lambda m: "%%%X" % ord(m.group(0)), text)
808def _sqla_url_unquote(text):
809 return util.unquote(text)
812def _parse_keyvalue_args(name):
813 m = re.match(r"(\w+)://(.*)", name)
814 if m is not None:
815 (name, args) = m.group(1, 2)
816 opts = dict(util.parse_qsl(args))
817 return URL(name, *opts)
818 else:
819 return None