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-2025 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, query_string: str, append: bool = False
368 ) -> URL:
369 """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
370 parameter dictionary updated by the given query string.
372 E.g.::
374 >>> from sqlalchemy.engine import make_url
375 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
376 >>> url = url.update_query_string(
377 ... "alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
378 ... )
379 >>> str(url)
380 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
382 :param query_string: a URL escaped query string, not including the
383 question mark.
385 :param append: if True, parameters in the existing query string will
386 not be removed; new parameters will be in addition to those present.
387 If left at its default of False, keys present in the given query
388 parameters will replace those of the existing query string.
390 .. versionadded:: 1.4
392 .. seealso::
394 :attr:`_engine.URL.query`
396 :meth:`_engine.URL.update_query_dict`
398 """ # noqa: E501
399 return self.update_query_pairs(parse_qsl(query_string), append=append)
401 def update_query_pairs(
402 self,
403 key_value_pairs: Iterable[Tuple[str, Union[str, List[str]]]],
404 append: bool = False,
405 ) -> URL:
406 """Return a new :class:`_engine.URL` object with the
407 :attr:`_engine.URL.query`
408 parameter dictionary updated by the given sequence of key/value pairs
410 E.g.::
412 >>> from sqlalchemy.engine import make_url
413 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
414 >>> url = url.update_query_pairs(
415 ... [
416 ... ("alt_host", "host1"),
417 ... ("alt_host", "host2"),
418 ... ("ssl_cipher", "/path/to/crt"),
419 ... ]
420 ... )
421 >>> str(url)
422 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
424 :param key_value_pairs: A sequence of tuples containing two strings
425 each.
427 :param append: if True, parameters in the existing query string will
428 not be removed; new parameters will be in addition to those present.
429 If left at its default of False, keys present in the given query
430 parameters will replace those of the existing query string.
432 .. versionadded:: 1.4
434 .. seealso::
436 :attr:`_engine.URL.query`
438 :meth:`_engine.URL.difference_update_query`
440 :meth:`_engine.URL.set`
442 """ # noqa: E501
444 existing_query = self.query
445 new_keys: Dict[str, Union[str, List[str]]] = {}
447 for key, value in key_value_pairs:
448 if key in new_keys:
449 new_keys[key] = util.to_list(new_keys[key])
450 cast("List[str]", new_keys[key]).append(cast(str, value))
451 else:
452 new_keys[key] = (
453 list(value) if isinstance(value, (list, tuple)) else value
454 )
456 new_query: Mapping[str, Union[str, Sequence[str]]]
457 if append:
458 new_query = {}
460 for k in new_keys:
461 if k in existing_query:
462 new_query[k] = tuple(
463 util.to_list(existing_query[k])
464 + util.to_list(new_keys[k])
465 )
466 else:
467 new_query[k] = new_keys[k]
469 new_query.update(
470 {
471 k: existing_query[k]
472 for k in set(existing_query).difference(new_keys)
473 }
474 )
475 else:
476 new_query = self.query.union(
477 {
478 k: tuple(v) if isinstance(v, list) else v
479 for k, v in new_keys.items()
480 }
481 )
482 return self.set(query=new_query)
484 def update_query_dict(
485 self,
486 query_parameters: Mapping[str, Union[str, List[str]]],
487 append: bool = False,
488 ) -> URL:
489 """Return a new :class:`_engine.URL` object with the
490 :attr:`_engine.URL.query` parameter dictionary updated by the given
491 dictionary.
493 The dictionary typically contains string keys and string values.
494 In order to represent a query parameter that is expressed multiple
495 times, pass a sequence of string values.
497 E.g.::
500 >>> from sqlalchemy.engine import make_url
501 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
502 >>> url = url.update_query_dict(
503 ... {"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"}
504 ... )
505 >>> str(url)
506 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
509 :param query_parameters: A dictionary with string keys and values
510 that are either strings, or sequences of strings.
512 :param append: if True, parameters in the existing query string will
513 not be removed; new parameters will be in addition to those present.
514 If left at its default of False, keys present in the given query
515 parameters will replace those of the existing query string.
518 .. versionadded:: 1.4
520 .. seealso::
522 :attr:`_engine.URL.query`
524 :meth:`_engine.URL.update_query_string`
526 :meth:`_engine.URL.update_query_pairs`
528 :meth:`_engine.URL.difference_update_query`
530 :meth:`_engine.URL.set`
532 """ # noqa: E501
533 return self.update_query_pairs(query_parameters.items(), append=append)
535 def difference_update_query(self, names: Iterable[str]) -> URL:
536 """
537 Remove the given names from the :attr:`_engine.URL.query` dictionary,
538 returning the new :class:`_engine.URL`.
540 E.g.::
542 url = url.difference_update_query(["foo", "bar"])
544 Equivalent to using :meth:`_engine.URL.set` as follows::
546 url = url.set(
547 query={
548 key: url.query[key]
549 for key in set(url.query).difference(["foo", "bar"])
550 }
551 )
553 .. versionadded:: 1.4
555 .. seealso::
557 :attr:`_engine.URL.query`
559 :meth:`_engine.URL.update_query_dict`
561 :meth:`_engine.URL.set`
563 """
565 if not set(names).intersection(self.query):
566 return self
568 return URL(
569 self.drivername,
570 self.username,
571 self.password,
572 self.host,
573 self.port,
574 self.database,
575 util.immutabledict(
576 {
577 key: self.query[key]
578 for key in set(self.query).difference(names)
579 }
580 ),
581 )
583 @property
584 def normalized_query(self) -> Mapping[str, Sequence[str]]:
585 """Return the :attr:`_engine.URL.query` dictionary with values normalized
586 into sequences.
588 As the :attr:`_engine.URL.query` dictionary may contain either
589 string values or sequences of string values to differentiate between
590 parameters that are specified multiple times in the query string,
591 code that needs to handle multiple parameters generically will wish
592 to use this attribute so that all parameters present are presented
593 as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
594 function. E.g.::
597 >>> from sqlalchemy.engine import make_url
598 >>> url = make_url(
599 ... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
600 ... )
601 >>> url.query
602 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
603 >>> url.normalized_query
604 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
606 """ # noqa: E501
608 return util.immutabledict(
609 {
610 k: (v,) if not isinstance(v, tuple) else v
611 for k, v in self.query.items()
612 }
613 )
615 @util.deprecated(
616 "1.4",
617 "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
618 "be removed in a future release. Please use the "
619 ":meth:`_engine.URL.render_as_string` method.",
620 )
621 def __to_string__(self, hide_password: bool = True) -> str:
622 """Render this :class:`_engine.URL` object as a string.
624 :param hide_password: Defaults to True. The password is not shown
625 in the string unless this is set to False.
627 """
628 return self.render_as_string(hide_password=hide_password)
630 def render_as_string(self, hide_password: bool = True) -> str:
631 """Render this :class:`_engine.URL` object as a string.
633 This method is used when the ``__str__()`` or ``__repr__()``
634 methods are used. The method directly includes additional options.
636 :param hide_password: Defaults to True. The password is not shown
637 in the string unless this is set to False.
639 """
640 s = self.drivername + "://"
641 if self.username is not None:
642 s += quote(self.username, safe=" +")
643 if self.password is not None:
644 s += ":" + (
645 "***"
646 if hide_password
647 else quote(str(self.password), safe=" +")
648 )
649 s += "@"
650 if self.host is not None:
651 if ":" in self.host:
652 s += f"[{self.host}]"
653 else:
654 s += self.host
655 if self.port is not None:
656 s += ":" + str(self.port)
657 if self.database is not None:
658 s += "/" + quote(self.database, safe=" +/")
659 if self.query:
660 keys = list(self.query)
661 keys.sort()
662 s += "?" + "&".join(
663 f"{quote_plus(k)}={quote_plus(element)}"
664 for k in keys
665 for element in util.to_list(self.query[k])
666 )
667 return s
669 def __repr__(self) -> str:
670 return self.render_as_string()
672 def __copy__(self) -> URL:
673 return self.__class__.create(
674 self.drivername,
675 self.username,
676 self.password,
677 self.host,
678 self.port,
679 self.database,
680 # note this is an immutabledict of str-> str / tuple of str,
681 # also fully immutable. does not require deepcopy
682 self.query,
683 )
685 def __deepcopy__(self, memo: Any) -> URL:
686 return self.__copy__()
688 def __hash__(self) -> int:
689 return hash(str(self))
691 def __eq__(self, other: Any) -> bool:
692 return (
693 isinstance(other, URL)
694 and self.drivername == other.drivername
695 and self.username == other.username
696 and self.password == other.password
697 and self.host == other.host
698 and self.database == other.database
699 and self.query == other.query
700 and self.port == other.port
701 )
703 def __ne__(self, other: Any) -> bool:
704 return not self == other
706 def get_backend_name(self) -> str:
707 """Return the backend name.
709 This is the name that corresponds to the database backend in
710 use, and is the portion of the :attr:`_engine.URL.drivername`
711 that is to the left of the plus sign.
713 """
714 if "+" not in self.drivername:
715 return self.drivername
716 else:
717 return self.drivername.split("+")[0]
719 def get_driver_name(self) -> str:
720 """Return the backend name.
722 This is the name that corresponds to the DBAPI driver in
723 use, and is the portion of the :attr:`_engine.URL.drivername`
724 that is to the right of the plus sign.
726 If the :attr:`_engine.URL.drivername` does not include a plus sign,
727 then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
728 is imported in order to get the driver name.
730 """
732 if "+" not in self.drivername:
733 return self.get_dialect().driver
734 else:
735 return self.drivername.split("+")[1]
737 def _instantiate_plugins(
738 self, kwargs: Mapping[str, Any]
739 ) -> Tuple[URL, List[Any], Dict[str, Any]]:
740 plugin_names = util.to_list(self.query.get("plugin", ()))
741 plugin_names += kwargs.get("plugins", [])
743 kwargs = dict(kwargs)
745 loaded_plugins = [
746 plugins.load(plugin_name)(self, kwargs)
747 for plugin_name in plugin_names
748 ]
750 u = self.difference_update_query(["plugin", "plugins"])
752 for plugin in loaded_plugins:
753 new_u = plugin.update_url(u)
754 if new_u is not None:
755 u = new_u
757 kwargs.pop("plugins", None)
759 return u, loaded_plugins, kwargs
761 def _get_entrypoint(self) -> Type[Dialect]:
762 """Return the "entry point" dialect class.
764 This is normally the dialect itself except in the case when the
765 returned class implements the get_dialect_cls() method.
767 """
768 if "+" not in self.drivername:
769 name = self.drivername
770 else:
771 name = self.drivername.replace("+", ".")
772 cls = registry.load(name)
773 # check for legacy dialects that
774 # would return a module with 'dialect' as the
775 # actual class
776 if (
777 hasattr(cls, "dialect")
778 and isinstance(cls.dialect, type)
779 and issubclass(cls.dialect, Dialect)
780 ):
781 return cls.dialect
782 else:
783 return cast("Type[Dialect]", cls)
785 def get_dialect(self, _is_async: bool = False) -> Type[Dialect]:
786 """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
787 to this URL's driver name.
789 """
790 entrypoint = self._get_entrypoint()
791 if _is_async:
792 dialect_cls = entrypoint.get_async_dialect_cls(self)
793 else:
794 dialect_cls = entrypoint.get_dialect_cls(self)
795 return dialect_cls
797 def translate_connect_args(
798 self, names: Optional[List[str]] = None, **kw: Any
799 ) -> Dict[str, Any]:
800 r"""Translate url attributes into a dictionary of connection arguments.
802 Returns attributes of this url (`host`, `database`, `username`,
803 `password`, `port`) as a plain dictionary. The attribute names are
804 used as the keys by default. Unset or false attributes are omitted
805 from the final dictionary.
807 :param \**kw: Optional, alternate key names for url attributes.
809 :param names: Deprecated. Same purpose as the keyword-based alternate
810 names, but correlates the name to the original positionally.
811 """
813 if names is not None:
814 util.warn_deprecated(
815 "The `URL.translate_connect_args.name`s parameter is "
816 "deprecated. Please pass the "
817 "alternate names as kw arguments.",
818 "1.4",
819 )
821 translated = {}
822 attribute_names = ["host", "database", "username", "password", "port"]
823 for sname in attribute_names:
824 if names:
825 name = names.pop(0)
826 elif sname in kw:
827 name = kw[sname]
828 else:
829 name = sname
830 if name is not None and getattr(self, sname, False):
831 if sname == "password":
832 translated[name] = str(getattr(self, sname))
833 else:
834 translated[name] = getattr(self, sname)
836 return translated
839def make_url(name_or_url: Union[str, URL]) -> URL:
840 """Given a string, produce a new URL instance.
842 The format of the URL generally follows `RFC-1738
843 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
844 that underscores, and not dashes or periods, are accepted within the
845 "scheme" portion.
847 If a :class:`.URL` object is passed, it is returned as is.
849 .. seealso::
851 :ref:`database_urls`
853 """
855 if isinstance(name_or_url, str):
856 return _parse_url(name_or_url)
857 elif not isinstance(name_or_url, URL) and not hasattr(
858 name_or_url, "_sqla_is_testing_if_this_is_a_mock_object"
859 ):
860 raise exc.ArgumentError(
861 f"Expected string or URL object, got {name_or_url!r}"
862 )
863 else:
864 return name_or_url
867def _parse_url(name: str) -> URL:
868 pattern = re.compile(
869 r"""
870 (?P<name>[\w\+]+)://
871 (?:
872 (?P<username>[^:/]*)
873 (?::(?P<password>[^@]*))?
874 @)?
875 (?:
876 (?:
877 \[(?P<ipv6host>[^/\?]+)\] |
878 (?P<ipv4host>[^/:\?]+)
879 )?
880 (?::(?P<port>[^/\?]*))?
881 )?
882 (?:/(?P<database>[^\?]*))?
883 (?:\?(?P<query>.*))?
884 """,
885 re.X,
886 )
888 m = pattern.match(name)
889 if m is not None:
890 components = m.groupdict()
891 query: Optional[Dict[str, Union[str, List[str]]]]
892 if components["query"] is not None:
893 query = {}
895 for key, value in parse_qsl(components["query"]):
896 if key in query:
897 query[key] = util.to_list(query[key])
898 cast("List[str]", query[key]).append(value)
899 else:
900 query[key] = value
901 else:
902 query = None
903 components["query"] = query
905 for comp in "username", "password", "database":
906 if components[comp] is not None:
907 components[comp] = unquote(components[comp])
909 ipv4host = components.pop("ipv4host")
910 ipv6host = components.pop("ipv6host")
911 components["host"] = ipv4host or ipv6host
912 name = components.pop("name")
914 if components["port"]:
915 components["port"] = int(components["port"])
917 return URL.create(name, **components) # type: ignore
919 else:
920 raise exc.ArgumentError(
921 "Could not parse SQLAlchemy URL from given URL string"
922 )