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 if not set(names).intersection(self.query):
580 return self
582 return URL(
583 self.drivername,
584 self.username,
585 self.password,
586 self.host,
587 self.port,
588 self.database,
589 util.immutabledict(
590 {
591 key: self.query[key]
592 for key in set(self.query).difference(names)
593 }
594 ),
595 )
597 @property
598 def normalized_query(self) -> Mapping[str, Sequence[str]]:
599 """Return the :attr:`_engine.URL.query` dictionary with values normalized
600 into sequences.
602 As the :attr:`_engine.URL.query` dictionary may contain either
603 string values or sequences of string values to differentiate between
604 parameters that are specified multiple times in the query string,
605 code that needs to handle multiple parameters generically will wish
606 to use this attribute so that all parameters present are presented
607 as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
608 function. E.g.::
611 >>> from sqlalchemy.engine import make_url
612 >>> url = make_url(
613 ... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
614 ... )
615 >>> url.query
616 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
617 >>> url.normalized_query
618 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
620 """ # noqa: E501
622 return util.immutabledict(
623 {
624 k: (v,) if not isinstance(v, tuple) else v
625 for k, v in self.query.items()
626 }
627 )
629 @util.deprecated(
630 "1.4",
631 "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
632 "be removed in a future release. Please use the "
633 ":meth:`_engine.URL.render_as_string` method.",
634 )
635 def __to_string__(self, hide_password: bool = True) -> str:
636 """Render this :class:`_engine.URL` object as a string.
638 :param hide_password: Defaults to True. The password is not shown
639 in the string unless this is set to False.
641 """
642 return self.render_as_string(hide_password=hide_password)
644 def render_as_string(self, hide_password: bool = True) -> str:
645 """Render this :class:`_engine.URL` object as a string.
647 This method is used when the ``__str__()`` or ``__repr__()``
648 methods are used. The method directly includes additional options.
650 :param hide_password: Defaults to True. The password is not shown
651 in the string unless this is set to False.
653 """
654 s = self.drivername + "://"
655 if self.username is not None:
656 s += quote(self.username, safe=" +")
657 if self.password is not None:
658 s += ":" + (
659 "***"
660 if hide_password
661 else quote(str(self.password), safe=" +")
662 )
663 s += "@"
664 if self.host is not None:
665 if ":" in self.host:
666 s += f"[{self.host}]"
667 else:
668 s += self.host
669 if self.port is not None:
670 s += ":" + str(self.port)
671 if self.database is not None:
672 s += "/" + quote(self.database, safe=" +/")
673 if self.query:
674 keys = list(self.query)
675 keys.sort()
676 s += "?" + "&".join(
677 f"{quote_plus(k)}={quote_plus(element)}"
678 for k in keys
679 for element in util.to_list(self.query[k])
680 )
681 return s
683 def __repr__(self) -> str:
684 return self.render_as_string()
686 def __copy__(self) -> URL:
687 return self.__class__.create(
688 self.drivername,
689 self.username,
690 self.password,
691 self.host,
692 self.port,
693 self.database,
694 # note this is an immutabledict of str-> str / tuple of str,
695 # also fully immutable. does not require deepcopy
696 self.query,
697 )
699 def __deepcopy__(self, memo: Any) -> URL:
700 return self.__copy__()
702 def __hash__(self) -> int:
703 return hash(str(self))
705 def __eq__(self, other: Any) -> bool:
706 return (
707 isinstance(other, URL)
708 and self.drivername == other.drivername
709 and self.username == other.username
710 and self.password == other.password
711 and self.host == other.host
712 and self.database == other.database
713 and self.query == other.query
714 and self.port == other.port
715 )
717 def __ne__(self, other: Any) -> bool:
718 return not self == other
720 def get_backend_name(self) -> str:
721 """Return the backend name.
723 This is the name that corresponds to the database backend in
724 use, and is the portion of the :attr:`_engine.URL.drivername`
725 that is to the left of the plus sign.
727 """
728 if "+" not in self.drivername:
729 return self.drivername
730 else:
731 return self.drivername.split("+")[0]
733 def get_driver_name(self) -> str:
734 """Return the backend name.
736 This is the name that corresponds to the DBAPI driver in
737 use, and is the portion of the :attr:`_engine.URL.drivername`
738 that is to the right of the plus sign.
740 If the :attr:`_engine.URL.drivername` does not include a plus sign,
741 then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
742 is imported in order to get the driver name.
744 """
746 if "+" not in self.drivername:
747 return self.get_dialect().driver
748 else:
749 return self.drivername.split("+")[1]
751 def _instantiate_plugins(
752 self, kwargs: Mapping[str, Any]
753 ) -> Tuple[URL, List[Any], Dict[str, Any]]:
754 plugin_names = util.to_list(self.query.get("plugin", ()))
755 plugin_names += kwargs.get("plugins", [])
757 kwargs = dict(kwargs)
759 loaded_plugins = [
760 plugins.load(plugin_name)(self, kwargs)
761 for plugin_name in plugin_names
762 ]
764 u = self.difference_update_query(["plugin", "plugins"])
766 for plugin in loaded_plugins:
767 new_u = plugin.update_url(u)
768 if new_u is not None:
769 u = new_u
771 kwargs.pop("plugins", None)
773 return u, loaded_plugins, kwargs
775 def _get_entrypoint(self) -> Type[Dialect]:
776 """Return the "entry point" dialect class.
778 This is normally the dialect itself except in the case when the
779 returned class implements the get_dialect_cls() method.
781 """
782 if "+" not in self.drivername:
783 name = self.drivername
784 else:
785 name = self.drivername.replace("+", ".")
786 cls = registry.load(name)
787 # check for legacy dialects that
788 # would return a module with 'dialect' as the
789 # actual class
790 if (
791 hasattr(cls, "dialect")
792 and isinstance(cls.dialect, type)
793 and issubclass(cls.dialect, Dialect)
794 ):
795 return cls.dialect
796 else:
797 return cast("Type[Dialect]", cls)
799 def get_dialect(self, _is_async: bool = False) -> Type[Dialect]:
800 """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
801 to this URL's driver name.
803 """
804 entrypoint = self._get_entrypoint()
805 if _is_async:
806 dialect_cls = entrypoint.get_async_dialect_cls(self)
807 else:
808 dialect_cls = entrypoint.get_dialect_cls(self)
809 return dialect_cls
811 def translate_connect_args(
812 self, names: Optional[List[str]] = None, **kw: Any
813 ) -> Dict[str, Any]:
814 r"""Translate url attributes into a dictionary of connection arguments.
816 Returns attributes of this url (`host`, `database`, `username`,
817 `password`, `port`) as a plain dictionary. The attribute names are
818 used as the keys by default. Unset or false attributes are omitted
819 from the final dictionary.
821 :param \**kw: Optional, alternate key names for url attributes.
823 :param names: Deprecated. Same purpose as the keyword-based alternate
824 names, but correlates the name to the original positionally.
825 """
827 if names is not None:
828 util.warn_deprecated(
829 "The `URL.translate_connect_args.name`s parameter is "
830 "deprecated. Please pass the "
831 "alternate names as kw arguments.",
832 "1.4",
833 )
835 translated = {}
836 attribute_names = ["host", "database", "username", "password", "port"]
837 for sname in attribute_names:
838 if names:
839 name = names.pop(0)
840 elif sname in kw:
841 name = kw[sname]
842 else:
843 name = sname
844 if name is not None and getattr(self, sname, False):
845 if sname == "password":
846 translated[name] = str(getattr(self, sname))
847 else:
848 translated[name] = getattr(self, sname)
850 return translated
853def make_url(
854 name_or_url: Union[str, URL], *, keep_blank_values: bool = False
855) -> URL:
856 """Given a string, produce a new URL instance.
858 The format of the URL generally follows `RFC-1738
859 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
860 that underscores, and not dashes or periods, are accepted within the
861 "scheme" portion.
863 If a :class:`.URL` object is passed, it is returned as is.
865 :param name_or_url: a string URL or :class:`.URL` object.
867 :param keep_blank_values: if True, query string parameters with blank
868 values such as ``key=`` are retained in the :attr:`.URL.query`
869 dictionary as empty strings. If left at its default of False,
870 parameters with blank values are omitted. Has no effect when a
871 :class:`.URL` object is passed.
873 As :func:`_sa.create_engine` parses string URLs using the default
874 setting, a string URL with blank values to be retained should be
875 parsed with :func:`.make_url` first, and the resulting :class:`.URL`
876 passed to :func:`_sa.create_engine`.
878 .. versionadded:: 2.1.6
880 .. seealso::
882 :ref:`database_urls`
884 """
886 if isinstance(name_or_url, str):
887 return _parse_url(name_or_url, keep_blank_values=keep_blank_values)
888 elif not isinstance(name_or_url, URL) and not hasattr(
889 name_or_url, "_sqla_is_testing_if_this_is_a_mock_object"
890 ):
891 raise exc.ArgumentError(
892 f"Expected string or URL object, got {name_or_url!r}"
893 )
894 else:
895 return name_or_url
898def _parse_url(name: str, keep_blank_values: bool = False) -> URL:
899 pattern = re.compile(
900 r"""
901 (?P<name>[\w\+]+)://
902 (?:
903 (?P<username>[^:/]*)
904 (?::(?P<password>[^@]*))?
905 @)?
906 (?:
907 (?:
908 \[(?P<ipv6host>[^/\?]+)\] |
909 (?P<ipv4host>[^/:\?]+)
910 )?
911 (?::(?P<port>[^/\?]*))?
912 )?
913 (?:/(?P<database>[^\?]*))?
914 (?:\?(?P<query>.*))?
915 """,
916 re.X,
917 )
919 m = pattern.match(name)
920 if m is not None:
921 components = m.groupdict()
922 query: Optional[Dict[str, Union[str, List[str]]]]
923 if components["query"] is not None:
924 query = {}
926 for key, value in parse_qsl(
927 components["query"], keep_blank_values=keep_blank_values
928 ):
929 if key in query:
930 query[key] = util.to_list(query[key])
931 cast("List[str]", query[key]).append(value)
932 else:
933 query[key] = value
934 else:
935 query = None
936 components["query"] = query
938 for comp in "username", "password", "database":
939 if components[comp] is not None:
940 components[comp] = unquote(components[comp])
942 ipv4host = components.pop("ipv4host")
943 ipv6host = components.pop("ipv6host")
944 components["host"] = ipv4host or ipv6host
945 name = components.pop("name")
947 if components["port"]:
948 components["port"] = int(components["port"])
950 return URL.create(name, **components) # type: ignore[arg-type]
952 else:
953 raise exc.ArgumentError(
954 "Could not parse SQLAlchemy URL from given URL string"
955 )