codekingpro/portable-devtools
115k
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
7
8"""Provides the :class:`~sqlalchemy.engine.url.URL` class which encapsulates
9information about a database connection specification.
10
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"""
16
17from __future__ import annotations
18
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
38
39from .interfaces import Dialect
40from .. import exc
41from .. import util
42from ..dialects import plugins
43from ..dialects import registry
44
45
46class URL(NamedTuple):
47 """
48 Represent the components of a URL used to connect to a database.
49
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.
57
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.
65
66 .. versionchanged:: 1.4
67
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`.
76
77 .. seealso::
78
79 :ref:`database_urls`
80
81 :class:`_engine.URL` contains the following attributes:
82
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.
93
94
95 """
96
97 drivername: str
98 """database backend and driver name, such as
99 ``postgresql+psycopg2``
100
101 """
102
103 username: Optional[str]
104 "username string"
105
106 password: Optional[str]
107 """password, which is normally a string but may also be any
108 object that has a ``__str__()`` method."""
109
110 host: Optional[str]
111 """hostname or IP number. May also be a data source name for some
112 drivers."""
113
114 port: Optional[int]
115 """integer port number"""
116
117 database: Optional[str]
118 """database name"""
119
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.::
123
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'})
130
131 To create a mutable copy of this mapping, use the ``dict`` constructor::
132
133 mutable_query_opts = dict(url.query)
134
135 .. seealso::
136
137 :attr:`_engine.URL.normalized_query` - normalizes all values into sequences
138 for consistent processing
139
140 Methods for altering the contents of :attr:`_engine.URL.query`:
141
142 :meth:`_engine.URL.update_query_dict`
143
144 :meth:`_engine.URL.update_query_string`
145
146 :meth:`_engine.URL.update_query_pairs`
147
148 :meth:`_engine.URL.difference_update_query`
149
150 """ # noqa: E501
151
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.
164
165 .. seealso::
166
167 :ref:`database_urls`
168
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()``.
175
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.
180
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`.
184
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.
196
197 .. versionadded:: 1.4
198
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.
205
206 """
207
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 )
217
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")
226
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
232
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
239
240 return cls._assert_str(v, paramname)
241
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
254
255 @overload
256 def _assert_value(
257 val: str,
258 ) -> str: ...
259
260 @overload
261 def _assert_value(
262 val: Sequence[str],
263 ) -> Union[str, Tuple[str, ...]]: ...
264
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 )
277
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
282
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()
288
289 return util.immutabledict(
290 {
291 _assert_str(key): _assert_value(
292 value,
293 )
294 for key, value in dict_items
295 }
296 )
297
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.
309
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``.
313
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.
322
323 :return: new :class:`_engine.URL` object.
324
325 .. versionadded:: 1.4
326
327 .. seealso::
328
329 :meth:`_engine.URL.update_query_dict`
330
331 """
332
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
348
349 return self._assert_replace(**kw)
350
351 def _assert_replace(self, **kw: Any) -> URL:
352 """argument checks before calling _replace()"""
353
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"])
363
364 return self._replace(**kw)
365
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.
371
372 E.g.::
373
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'
381
382 :param query_string: a URL escaped query string, not including the
383 question mark.
384
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.
389
390 .. versionadded:: 1.4
391
392 .. seealso::
393
394 :attr:`_engine.URL.query`
395
396 :meth:`_engine.URL.update_query_dict`
397
398 """ # noqa: E501
399 return self.update_query_pairs(parse_qsl(query_string), append=append)
400
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
409
410 E.g.::
411
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'
423
424 :param key_value_pairs: A sequence of tuples containing two strings
425 each.
426
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.
431
432 .. versionadded:: 1.4
433
434 .. seealso::
435
436 :attr:`_engine.URL.query`
437
438 :meth:`_engine.URL.difference_update_query`
439
440 :meth:`_engine.URL.set`
441
442 """ # noqa: E501
443
444 existing_query = self.query
445 new_keys: Dict[str, Union[str, List[str]]] = {}
446
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 )
455
456 new_query: Mapping[str, Union[str, Sequence[str]]]
457 if append:
458 new_query = {}
459
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]
468
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)
483
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.
492
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.
496
497 E.g.::
498
499
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'
507
508
509 :param query_parameters: A dictionary with string keys and values
510 that are either strings, or sequences of strings.
511
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.
516
517
518 .. versionadded:: 1.4
519
520 .. seealso::
521
522 :attr:`_engine.URL.query`
523
524 :meth:`_engine.URL.update_query_string`
525
526 :meth:`_engine.URL.update_query_pairs`
527
528 :meth:`_engine.URL.difference_update_query`
529
530 :meth:`_engine.URL.set`
531
532 """ # noqa: E501
533 return self.update_query_pairs(query_parameters.items(), append=append)
534
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`.
539
540 E.g.::
541
542 url = url.difference_update_query(["foo", "bar"])
543
544 Equivalent to using :meth:`_engine.URL.set` as follows::
545
546 url = url.set(
547 query={
548 key: url.query[key]
549 for key in set(url.query).difference(["foo", "bar"])
550 }
551 )
552
553 .. versionadded:: 1.4
554
555 .. seealso::
556
557 :attr:`_engine.URL.query`
558
559 :meth:`_engine.URL.update_query_dict`
560
561 :meth:`_engine.URL.set`
562
563 """
564
565 if not set(names).intersection(self.query):
566 return self
567
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 )
582
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.
587
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.::
595
596
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',)})
605
606 """ # noqa: E501
607
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 )
614
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.
623
624 :param hide_password: Defaults to True. The password is not shown
625 in the string unless this is set to False.
626
627 """
628 return self.render_as_string(hide_password=hide_password)
629
630 def render_as_string(self, hide_password: bool = True) -> str:
631 """Render this :class:`_engine.URL` object as a string.
632
633 This method is used when the ``__str__()`` or ``__repr__()``
634 methods are used. The method directly includes additional options.
635
636 :param hide_password: Defaults to True. The password is not shown
637 in the string unless this is set to False.
638
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 += "/" + self.database
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
668
669 def __repr__(self) -> str:
670 return self.render_as_string()
671
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 )
684
685 def __deepcopy__(self, memo: Any) -> URL:
686 return self.__copy__()
687
688 def __hash__(self) -> int:
689 return hash(str(self))
690
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 )
702
703 def __ne__(self, other: Any) -> bool:
704 return not self == other
705
706 def get_backend_name(self) -> str:
707 """Return the backend name.
708
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.
712
713 """
714 if "+" not in self.drivername:
715 return self.drivername
716 else:
717 return self.drivername.split("+")[0]
718
719 def get_driver_name(self) -> str:
720 """Return the backend name.
721
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.
725
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.
729
730 """
731
732 if "+" not in self.drivername:
733 return self.get_dialect().driver
734 else:
735 return self.drivername.split("+")[1]
736
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", [])
742
743 kwargs = dict(kwargs)
744
745 loaded_plugins = [
746 plugins.load(plugin_name)(self, kwargs)
747 for plugin_name in plugin_names
748 ]
749
750 u = self.difference_update_query(["plugin", "plugins"])
751
752 for plugin in loaded_plugins:
753 new_u = plugin.update_url(u)
754 if new_u is not None:
755 u = new_u
756
757 kwargs.pop("plugins", None)
758
759 return u, loaded_plugins, kwargs
760
761 def _get_entrypoint(self) -> Type[Dialect]:
762 """Return the "entry point" dialect class.
763
764 This is normally the dialect itself except in the case when the
765 returned class implements the get_dialect_cls() method.
766
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)
784
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.
788
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
796
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.
801
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.
806
807 :param \**kw: Optional, alternate key names for url attributes.
808
809 :param names: Deprecated. Same purpose as the keyword-based alternate
810 names, but correlates the name to the original positionally.
811 """
812
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 )
820
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)
835
836 return translated
837
838
839def make_url(name_or_url: Union[str, URL]) -> URL:
840 """Given a string, produce a new URL instance.
841
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.
846
847 If a :class:`.URL` object is passed, it is returned as is.
848
849 .. seealso::
850
851 :ref:`database_urls`
852
853 """
854
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
865
866
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 )
887
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 = {}
894
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
904
905 if components["username"] is not None:
906 components["username"] = unquote(components["username"])
907
908 if components["password"] is not None:
909 components["password"] = unquote(components["password"])
910
911 ipv4host = components.pop("ipv4host")
912 ipv6host = components.pop("ipv6host")
913 components["host"] = ipv4host or ipv6host
914 name = components.pop("name")
915
916 if components["port"]:
917 components["port"] = int(components["port"])
918
919 return URL.create(name, **components) # type: ignore
920
921 else:
922 raise exc.ArgumentError(
923 "Could not parse SQLAlchemy URL from given URL string"
924 )
925