Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# engine/url.py
2# Copyright (C) 2005-2026 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"""
17from __future__ import annotations
19import collections.abc as collections_abc
20import re
21from typing import Any
22from typing import cast
23from typing import Dict
24from typing import Iterable
25from typing import List
26from typing import Mapping
27from typing import NamedTuple
28from typing import Optional
29from typing import overload
30from typing import Sequence
31from typing import Tuple
32from typing import Type
33from typing import Union
34from urllib.parse import parse_qsl
35from urllib.parse import quote
36from urllib.parse import quote_plus
37from urllib.parse import unquote
39from .interfaces import Dialect
40from .. import exc
41from .. import util
42from ..dialects import plugins
43from ..dialects import registry
46class URL(NamedTuple):
47 """
48 Represent the components of a URL used to connect to a database.
50 URLs are typically constructed from a fully formatted URL string, where the
51 :func:`.make_url` function is used internally by the
52 :func:`_sa.create_engine` function in order to parse the URL string into
53 its individual components, which are then used to construct a new
54 :class:`.URL` object. When parsing from a formatted URL string, the parsing
55 format generally follows
56 `RFC-1738 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions.
58 A :class:`_engine.URL` object may also be produced directly, either by
59 using the :func:`.make_url` function with a fully formed URL string, or
60 by using the :meth:`_engine.URL.create` constructor in order
61 to construct a :class:`_engine.URL` programmatically given individual
62 fields. The resulting :class:`.URL` object may be passed directly to
63 :func:`_sa.create_engine` in place of a string argument, which will bypass
64 the usage of :func:`.make_url` within the engine's creation process.
66 .. versionchanged:: 1.4
68 The :class:`_engine.URL` object is now an immutable object. To
69 create a URL, use the :func:`_engine.make_url` or
70 :meth:`_engine.URL.create` function / method. To modify
71 a :class:`_engine.URL`, use methods like
72 :meth:`_engine.URL.set` and
73 :meth:`_engine.URL.update_query_dict` to return a new
74 :class:`_engine.URL` object with modifications. See notes for this
75 change at :ref:`change_5526`.
77 .. seealso::
79 :ref:`database_urls`
81 :class:`_engine.URL` contains the following attributes:
83 * :attr:`_engine.URL.drivername`: database backend and driver name, such as
84 ``postgresql+psycopg2``
85 * :attr:`_engine.URL.username`: username string
86 * :attr:`_engine.URL.password`: password string
87 * :attr:`_engine.URL.host`: string hostname
88 * :attr:`_engine.URL.port`: integer port number
89 * :attr:`_engine.URL.database`: string database name
90 * :attr:`_engine.URL.query`: an immutable mapping representing the query
91 string. contains strings for keys and either strings or tuples of
92 strings for values.
95 """
97 drivername: str
98 """database backend and driver name, such as
99 ``postgresql+psycopg2``
101 """
103 username: Optional[str]
104 "username string"
106 password: Optional[str]
107 """password, which is normally a string but may also be any
108 object that has a ``__str__()`` method."""
110 host: Optional[str]
111 """hostname or IP number. May also be a data source name for some
112 drivers."""
114 port: Optional[int]
115 """integer port number"""
117 database: Optional[str]
118 """database name"""
120 query: util.immutabledict[str, Union[Tuple[str, ...], str]]
121 """an immutable mapping representing the query string. contains strings
122 for keys and either strings or tuples of strings for values, e.g.::
124 >>> from sqlalchemy.engine import make_url
125 >>> url = make_url(
126 ... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
127 ... )
128 >>> url.query
129 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
131 To create a mutable copy of this mapping, use the ``dict`` constructor::
133 mutable_query_opts = dict(url.query)
135 .. seealso::
137 :attr:`_engine.URL.normalized_query` - normalizes all values into sequences
138 for consistent processing
140 Methods for altering the contents of :attr:`_engine.URL.query`:
142 :meth:`_engine.URL.update_query_dict`
144 :meth:`_engine.URL.update_query_string`
146 :meth:`_engine.URL.update_query_pairs`
148 :meth:`_engine.URL.difference_update_query`
150 """ # noqa: E501
152 @classmethod
153 def create(
154 cls,
155 drivername: str,
156 username: Optional[str] = None,
157 password: Optional[str] = None,
158 host: Optional[str] = None,
159 port: Optional[int] = None,
160 database: Optional[str] = None,
161 query: Mapping[str, Union[Sequence[str], str]] = util.EMPTY_DICT,
162 ) -> URL:
163 """Create a new :class:`_engine.URL` object.
165 .. seealso::
167 :ref:`database_urls`
169 :param drivername: the name of the database backend. This name will
170 correspond to a module in sqlalchemy/databases or a third party
171 plug-in.
172 :param username: The user name.
173 :param password: database password. Is typically a string, but may
174 also be an object that can be stringified with ``str()``.
176 .. note:: The password string should **not** be URL encoded when
177 passed as an argument to :meth:`_engine.URL.create`; the string
178 should contain the password characters exactly as they would be
179 typed.
181 .. note:: A password-producing object will be stringified only
182 **once** per :class:`_engine.Engine` object. For dynamic password
183 generation per connect, see :ref:`engines_dynamic_tokens`.
185 :param host: The name of the host.
186 :param port: The port number.
187 :param database: The database name.
188 :param query: A dictionary of string keys to string values to be passed
189 to the dialect and/or the DBAPI upon connect. To specify non-string
190 parameters to a Python DBAPI directly, use the
191 :paramref:`_sa.create_engine.connect_args` parameter to
192 :func:`_sa.create_engine`. See also
193 :attr:`_engine.URL.normalized_query` for a dictionary that is
194 consistently string->list of string.
195 :return: new :class:`_engine.URL` object.
197 .. versionadded:: 1.4
199 The :class:`_engine.URL` object is now an **immutable named
200 tuple**. In addition, the ``query`` dictionary is also immutable.
201 To create a URL, use the :func:`_engine.url.make_url` or
202 :meth:`_engine.URL.create` function/ method. To modify a
203 :class:`_engine.URL`, use the :meth:`_engine.URL.set` and
204 :meth:`_engine.URL.update_query` methods.
206 """
208 return cls(
209 cls._assert_str(drivername, "drivername"),
210 cls._assert_none_str(username, "username"),
211 password,
212 cls._assert_none_str(host, "host"),
213 cls._assert_port(port),
214 cls._assert_none_str(database, "database"),
215 cls._str_dict(query),
216 )
218 @classmethod
219 def _assert_port(cls, port: Optional[int]) -> Optional[int]:
220 if port is None:
221 return None
222 try:
223 return int(port)
224 except TypeError:
225 raise TypeError("Port argument must be an integer or None")
227 @classmethod
228 def _assert_str(cls, v: str, paramname: str) -> str:
229 if not isinstance(v, str):
230 raise TypeError("%s must be a string" % paramname)
231 return v
233 @classmethod
234 def _assert_none_str(
235 cls, v: Optional[str], paramname: str
236 ) -> Optional[str]:
237 if v is None:
238 return v
240 return cls._assert_str(v, paramname)
242 @classmethod
243 def _str_dict(
244 cls,
245 dict_: Optional[
246 Union[
247 Sequence[Tuple[str, Union[Sequence[str], str]]],
248 Mapping[str, Union[Sequence[str], str]],
249 ]
250 ],
251 ) -> util.immutabledict[str, Union[Tuple[str, ...], str]]:
252 if dict_ is None:
253 return util.EMPTY_DICT
255 @overload
256 def _assert_value(
257 val: str,
258 ) -> str: ...
260 @overload
261 def _assert_value(
262 val: Sequence[str],
263 ) -> Union[str, Tuple[str, ...]]: ...
265 def _assert_value(
266 val: Union[str, Sequence[str]],
267 ) -> Union[str, Tuple[str, ...]]:
268 if isinstance(val, str):
269 return val
270 elif isinstance(val, collections_abc.Sequence):
271 return tuple(_assert_value(elem) for elem in val)
272 else:
273 raise TypeError(
274 "Query dictionary values must be strings or "
275 "sequences of strings"
276 )
278 def _assert_str(v: str) -> str:
279 if not isinstance(v, str):
280 raise TypeError("Query dictionary keys must be strings")
281 return v
283 dict_items: Iterable[Tuple[str, Union[Sequence[str], str]]]
284 if isinstance(dict_, collections_abc.Sequence):
285 dict_items = dict_
286 else:
287 dict_items = dict_.items()
289 return util.immutabledict(
290 {
291 _assert_str(key): _assert_value(
292 value,
293 )
294 for key, value in dict_items
295 }
296 )
298 def set(
299 self,
300 drivername: Optional[str] = None,
301 username: Optional[str] = None,
302 password: Optional[str] = None,
303 host: Optional[str] = None,
304 port: Optional[int] = None,
305 database: Optional[str] = None,
306 query: Optional[Mapping[str, Union[Sequence[str], str]]] = None,
307 ) -> URL:
308 """return a new :class:`_engine.URL` object with modifications.
310 Values are used if they are non-None. To set a value to ``None``
311 explicitly, use the :meth:`_engine.URL._replace` method adapted
312 from ``namedtuple``.
314 :param drivername: new drivername
315 :param username: new username
316 :param password: new password
317 :param host: new hostname
318 :param port: new port
319 :param query: new query parameters, passed a dict of string keys
320 referring to string or sequence of string values. Fully
321 replaces the previous list of arguments.
323 :return: new :class:`_engine.URL` object.
325 .. versionadded:: 1.4
327 .. seealso::
329 :meth:`_engine.URL.update_query_dict`
331 """
333 kw: Dict[str, Any] = {}
334 if drivername is not None:
335 kw["drivername"] = drivername
336 if username is not None:
337 kw["username"] = username
338 if password is not None:
339 kw["password"] = password
340 if host is not None:
341 kw["host"] = host
342 if port is not None:
343 kw["port"] = port
344 if database is not None:
345 kw["database"] = database
346 if query is not None:
347 kw["query"] = query
349 return self._assert_replace(**kw)
351 def _assert_replace(self, **kw: Any) -> URL:
352 """argument checks before calling _replace()"""
354 if "drivername" in kw:
355 self._assert_str(kw["drivername"], "drivername")
356 for name in "username", "host", "database":
357 if name in kw:
358 self._assert_none_str(kw[name], name)
359 if "port" in kw:
360 self._assert_port(kw["port"])
361 if "query" in kw:
362 kw["query"] = self._str_dict(kw["query"])
364 return self._replace(**kw)
366 def update_query_string(
367 self,
368 query_string: str,
369 append: bool = False,
370 *,
371 keep_blank_values: bool = False,
372 ) -> URL:
373 """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
374 parameter dictionary updated by the given query string.
376 E.g.::
378 >>> from sqlalchemy.engine import make_url
379 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
380 >>> url = url.update_query_string(
381 ... "alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
382 ... )
383 >>> str(url)
384 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
386 :param query_string: a URL escaped query string, not including the
387 question mark.
389 :param append: if True, parameters in the existing query string will
390 not be removed; new parameters will be in addition to those present.
391 If left at its default of False, keys present in the given query
392 parameters will replace those of the existing query string.
394 :param keep_blank_values: if True, parameters with blank values
395 such as ``key=`` are retained in the query dictionary as empty
396 strings. If left at its default of False, parameters with blank
397 values are omitted.
399 .. versionadded:: 2.1.6
401 .. versionadded:: 1.4
403 .. seealso::
405 :attr:`_engine.URL.query`
407 :meth:`_engine.URL.update_query_dict`
409 """ # noqa: E501
410 return self.update_query_pairs(
411 parse_qsl(query_string, keep_blank_values=keep_blank_values),
412 append=append,
413 )
415 def update_query_pairs(
416 self,
417 key_value_pairs: Iterable[Tuple[str, Union[str, List[str]]]],
418 append: bool = False,
419 ) -> URL:
420 """Return a new :class:`_engine.URL` object with the
421 :attr:`_engine.URL.query`
422 parameter dictionary updated by the given sequence of key/value pairs
424 E.g.::
426 >>> from sqlalchemy.engine import make_url
427 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
428 >>> url = url.update_query_pairs(
429 ... [
430 ... ("alt_host", "host1"),
431 ... ("alt_host", "host2"),
432 ... ("ssl_cipher", "/path/to/crt"),
433 ... ]
434 ... )
435 >>> str(url)
436 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
438 :param key_value_pairs: A sequence of tuples containing two strings
439 each.
441 :param append: if True, parameters in the existing query string will
442 not be removed; new parameters will be in addition to those present.
443 If left at its default of False, keys present in the given query
444 parameters will replace those of the existing query string.
446 .. versionadded:: 1.4
448 .. seealso::
450 :attr:`_engine.URL.query`
452 :meth:`_engine.URL.difference_update_query`
454 :meth:`_engine.URL.set`
456 """ # noqa: E501
458 existing_query = self.query
459 new_keys: Dict[str, Union[str, List[str]]] = {}
461 for key, value in key_value_pairs:
462 if key in new_keys:
463 new_keys[key] = util.to_list(new_keys[key])
464 cast("List[str]", new_keys[key]).append(cast(str, value))
465 else:
466 new_keys[key] = (
467 list(value) if isinstance(value, (list, tuple)) else value
468 )
470 new_query: Mapping[str, Union[str, Sequence[str]]]
471 if append:
472 new_query = {}
474 for k in new_keys:
475 if k in existing_query:
476 new_query[k] = tuple(
477 util.to_list(existing_query[k])
478 + util.to_list(new_keys[k])
479 )
480 else:
481 new_query[k] = new_keys[k]
483 new_query.update(
484 {
485 k: existing_query[k]
486 for k in set(existing_query).difference(new_keys)
487 }
488 )
489 else:
490 new_query = self.query.union(
491 {
492 k: tuple(v) if isinstance(v, list) else v
493 for k, v in new_keys.items()
494 }
495 )
496 return self.set(query=new_query)
498 def update_query_dict(
499 self,
500 query_parameters: Mapping[str, Union[str, List[str]]],
501 append: bool = False,
502 ) -> URL:
503 """Return a new :class:`_engine.URL` object with the
504 :attr:`_engine.URL.query` parameter dictionary updated by the given
505 dictionary.
507 The dictionary typically contains string keys and string values.
508 In order to represent a query parameter that is expressed multiple
509 times, pass a sequence of string values.
511 E.g.::
514 >>> from sqlalchemy.engine import make_url
515 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
516 >>> url = url.update_query_dict(
517 ... {"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"}
518 ... )
519 >>> str(url)
520 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
523 :param query_parameters: A dictionary with string keys and values
524 that are either strings, or sequences of strings.
526 :param append: if True, parameters in the existing query string will
527 not be removed; new parameters will be in addition to those present.
528 If left at its default of False, keys present in the given query
529 parameters will replace those of the existing query string.
532 .. versionadded:: 1.4
534 .. seealso::
536 :attr:`_engine.URL.query`
538 :meth:`_engine.URL.update_query_string`
540 :meth:`_engine.URL.update_query_pairs`
542 :meth:`_engine.URL.difference_update_query`
544 :meth:`_engine.URL.set`
546 """ # noqa: E501
547 return self.update_query_pairs(query_parameters.items(), append=append)
549 def difference_update_query(self, names: Iterable[str]) -> URL:
550 """
551 Remove the given names from the :attr:`_engine.URL.query` dictionary,
552 returning the new :class:`_engine.URL`.
554 E.g.::
556 url = url.difference_update_query(["foo", "bar"])
558 Equivalent to using :meth:`_engine.URL.set` as follows::
560 url = url.set(
561 query={
562 key: url.query[key]
563 for key in set(url.query).difference(["foo", "bar"])
564 }
565 )
567 .. versionadded:: 1.4
569 .. seealso::
571 :attr:`_engine.URL.query`
573 :meth:`_engine.URL.update_query_dict`
575 :meth:`_engine.URL.set`
577 """
579 names = set(names)
580 if not names.intersection(self.query):
581 return self
583 return URL(
584 self.drivername,
585 self.username,
586 self.password,
587 self.host,
588 self.port,
589 self.database,
590 util.immutabledict(
591 {
592 key: self.query[key]
593 for key in set(self.query).difference(names)
594 }
595 ),
596 )
598 @property
599 def normalized_query(self) -> Mapping[str, Sequence[str]]:
600 """Return the :attr:`_engine.URL.query` dictionary with values normalized
601 into sequences.
603 As the :attr:`_engine.URL.query` dictionary may contain either
604 string values or sequences of string values to differentiate between
605 parameters that are specified multiple times in the query string,
606 code that needs to handle multiple parameters generically will wish
607 to use this attribute so that all parameters present are presented
608 as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
609 function. E.g.::
612 >>> from sqlalchemy.engine import make_url
613 >>> url = make_url(
614 ... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
615 ... )
616 >>> url.query
617 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
618 >>> url.normalized_query
619 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
621 """ # noqa: E501
623 return util.immutabledict(
624 {
625 k: (v,) if not isinstance(v, tuple) else v
626 for k, v in self.query.items()
627 }
628 )
630 @util.deprecated(
631 "1.4",
632 "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
633 "be removed in a future release. Please use the "
634 ":meth:`_engine.URL.render_as_string` method.",
635 )
636 def __to_string__(self, hide_password: bool = True) -> str:
637 """Render this :class:`_engine.URL` object as a string.
639 :param hide_password: Defaults to True. The password is not shown
640 in the string unless this is set to False.
642 """
643 return self.render_as_string(hide_password=hide_password)
645 def render_as_string(self, hide_password: bool = True) -> str:
646 """Render this :class:`_engine.URL` object as a string.
648 This method is used when the ``__str__()`` or ``__repr__()``
649 methods are used. The method directly includes additional options.
651 :param hide_password: Defaults to True. The password is not shown
652 in the string unless this is set to False.
654 """
655 s = self.drivername + "://"
656 if self.username is not None:
657 s += quote(self.username, safe=" +")
658 if self.password is not None:
659 s += ":" + (
660 "***"
661 if hide_password
662 else quote(str(self.password), safe=" +")
663 )
664 s += "@"
665 if self.host is not None:
666 if ":" in self.host:
667 s += f"[{self.host}]"
668 else:
669 s += self.host
670 if self.port is not None:
671 s += ":" + str(self.port)
672 if self.database is not None:
673 s += "/" + quote(self.database, safe=" +/")
674 if self.query:
675 keys = list(self.query)
676 keys.sort()
677 s += "?" + "&".join(
678 f"{quote_plus(k)}={quote_plus(element)}"
679 for k in keys
680 for element in util.to_list(self.query[k])
681 )
682 return s
684 def __repr__(self) -> str:
685 return self.render_as_string()
687 def __copy__(self) -> URL:
688 return self.__class__.create(
689 self.drivername,
690 self.username,
691 self.password,
692 self.host,
693 self.port,
694 self.database,
695 # note this is an immutabledict of str-> str / tuple of str,
696 # also fully immutable. does not require deepcopy
697 self.query,
698 )
700 def __deepcopy__(self, memo: Any) -> URL:
701 return self.__copy__()
703 def __hash__(self) -> int:
704 return hash(str(self))
706 def __eq__(self, other: Any) -> bool:
707 return (
708 isinstance(other, URL)
709 and self.drivername == other.drivername
710 and self.username == other.username
711 and self.password == other.password
712 and self.host == other.host
713 and self.database == other.database
714 and self.query == other.query
715 and self.port == other.port
716 )
718 def __ne__(self, other: Any) -> bool:
719 return not self == other
721 def get_backend_name(self) -> str:
722 """Return the backend name.
724 This is the name that corresponds to the database backend in
725 use, and is the portion of the :attr:`_engine.URL.drivername`
726 that is to the left of the plus sign.
728 """
729 if "+" not in self.drivername:
730 return self.drivername
731 else:
732 return self.drivername.split("+")[0]
734 def get_driver_name(self) -> str:
735 """Return the backend name.
737 This is the name that corresponds to the DBAPI driver in
738 use, and is the portion of the :attr:`_engine.URL.drivername`
739 that is to the right of the plus sign.
741 If the :attr:`_engine.URL.drivername` does not include a plus sign,
742 then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
743 is imported in order to get the driver name.
745 """
747 if "+" not in self.drivername:
748 return self.get_dialect().driver
749 else:
750 return self.drivername.split("+")[1]
752 def _instantiate_plugins(
753 self, kwargs: Mapping[str, Any]
754 ) -> Tuple[URL, List[Any], Dict[str, Any]]:
755 plugin_names = util.to_list(self.query.get("plugin", ()))
756 plugin_names += kwargs.get("plugins", [])
758 kwargs = dict(kwargs)
760 loaded_plugins = [
761 plugins.load(plugin_name)(self, kwargs)
762 for plugin_name in plugin_names
763 ]
765 u = self.difference_update_query(["plugin", "plugins"])
767 for plugin in loaded_plugins:
768 new_u = plugin.update_url(u)
769 if new_u is not None:
770 u = new_u
772 kwargs.pop("plugins", None)
774 return u, loaded_plugins, kwargs
776 def _get_entrypoint(self) -> Type[Dialect]:
777 """Return the "entry point" dialect class.
779 This is normally the dialect itself except in the case when the
780 returned class implements the get_dialect_cls() method.
782 """
783 if "+" not in self.drivername:
784 name = self.drivername
785 else:
786 name = self.drivername.replace("+", ".")
787 cls = registry.load(name)
788 # check for legacy dialects that
789 # would return a module with 'dialect' as the
790 # actual class
791 if (
792 hasattr(cls, "dialect")
793 and isinstance(cls.dialect, type)
794 and issubclass(cls.dialect, Dialect)
795 ):
796 return cls.dialect
797 else:
798 return cast("Type[Dialect]", cls)
800 def get_dialect(self, _is_async: bool = False) -> Type[Dialect]:
801 """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
802 to this URL's driver name.
804 """
805 entrypoint = self._get_entrypoint()
806 if _is_async:
807 dialect_cls = entrypoint.get_async_dialect_cls(self)
808 else:
809 dialect_cls = entrypoint.get_dialect_cls(self)
810 return dialect_cls
812 def translate_connect_args(
813 self, names: Optional[List[str]] = None, **kw: Any
814 ) -> Dict[str, Any]:
815 r"""Translate url attributes into a dictionary of connection arguments.
817 Returns attributes of this url (`host`, `database`, `username`,
818 `password`, `port`) as a plain dictionary. The attribute names are
819 used as the keys by default. Unset or false attributes are omitted
820 from the final dictionary.
822 :param \**kw: Optional, alternate key names for url attributes.
824 :param names: Deprecated. Same purpose as the keyword-based alternate
825 names, but correlates the name to the original positionally.
826 """
828 if names is not None:
829 util.warn_deprecated(
830 "The `URL.translate_connect_args.name`s parameter is "
831 "deprecated. Please pass the "
832 "alternate names as kw arguments.",
833 "1.4",
834 )
836 translated = {}
837 attribute_names = ["host", "database", "username", "password", "port"]
838 for sname in attribute_names:
839 if names:
840 name = names.pop(0)
841 elif sname in kw:
842 name = kw[sname]
843 else:
844 name = sname
845 if name is not None and getattr(self, sname, False):
846 if sname == "password":
847 translated[name] = str(getattr(self, sname))
848 else:
849 translated[name] = getattr(self, sname)
851 return translated
854def make_url(
855 name_or_url: Union[str, URL], *, keep_blank_values: bool = False
856) -> URL:
857 """Given a string, produce a new URL instance.
859 The format of the URL generally follows `RFC-1738
860 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
861 that underscores, and not dashes or periods, are accepted within the
862 "scheme" portion.
864 If a :class:`.URL` object is passed, it is returned as is.
866 :param name_or_url: a string URL or :class:`.URL` object.
868 :param keep_blank_values: if True, query string parameters with blank
869 values such as ``key=`` are retained in the :attr:`.URL.query`
870 dictionary as empty strings. If left at its default of False,
871 parameters with blank values are omitted. Has no effect when a
872 :class:`.URL` object is passed.
874 As :func:`_sa.create_engine` parses string URLs using the default
875 setting, a string URL with blank values to be retained should be
876 parsed with :func:`.make_url` first, and the resulting :class:`.URL`
877 passed to :func:`_sa.create_engine`.
879 .. versionadded:: 2.1.6
881 .. seealso::
883 :ref:`database_urls`
885 """
887 if isinstance(name_or_url, str):
888 return _parse_url(name_or_url, keep_blank_values=keep_blank_values)
889 elif not isinstance(name_or_url, URL) and not hasattr(
890 name_or_url, "_sqla_is_testing_if_this_is_a_mock_object"
891 ):
892 raise exc.ArgumentError(
893 f"Expected string or URL object, got {name_or_url!r}"
894 )
895 else:
896 return name_or_url
899def _parse_url(name: str, keep_blank_values: bool = False) -> URL:
900 pattern = re.compile(
901 r"""
902 (?P<name>[\w\+]+)://
903 (?:
904 (?P<username>[^:/]*)
905 (?::(?P<password>[^@]*))?
906 @)?
907 (?:
908 (?:
909 \[(?P<ipv6host>[^/\?]+)\] |
910 (?P<ipv4host>[^/:\?]+)
911 )?
912 (?::(?P<port>[^/\?]*))?
913 )?
914 (?:/(?P<database>[^\?]*))?
915 (?:\?(?P<query>.*))?
916 """,
917 re.X,
918 )
920 m = pattern.match(name)
921 if m is not None:
922 components = m.groupdict()
923 query: Optional[Dict[str, Union[str, List[str]]]]
924 if components["query"] is not None:
925 query = {}
927 for key, value in parse_qsl(
928 components["query"], keep_blank_values=keep_blank_values
929 ):
930 if key in query:
931 query[key] = util.to_list(query[key])
932 cast("List[str]", query[key]).append(value)
933 else:
934 query[key] = value
935 else:
936 query = None
937 components["query"] = query
939 for comp in "username", "password", "database":
940 if components[comp] is not None:
941 components[comp] = unquote(components[comp])
943 ipv4host = components.pop("ipv4host")
944 ipv6host = components.pop("ipv6host")
945 components["host"] = ipv4host or ipv6host
946 name = components.pop("name")
948 if components["port"]:
949 components["port"] = int(components["port"])
951 return URL.create(name, **components) # type: ignore[arg-type]
953 else:
954 raise exc.ArgumentError(
955 "Could not parse SQLAlchemy URL from given URL string"
956 )