codekingpro/portable-devtools
114k
1# engine/url.py
2# Copyright (C) 2005-2024 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("postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
126 >>> url.query
127 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
128
129 To create a mutable copy of this mapping, use the ``dict`` constructor::
130
131 mutable_query_opts = dict(url.query)
132
133 .. seealso::
134
135 :attr:`_engine.URL.normalized_query` - normalizes all values into sequences
136 for consistent processing
137
138 Methods for altering the contents of :attr:`_engine.URL.query`:
139
140 :meth:`_engine.URL.update_query_dict`
141
142 :meth:`_engine.URL.update_query_string`
143
144 :meth:`_engine.URL.update_query_pairs`
145
146 :meth:`_engine.URL.difference_update_query`
147
148 """ # noqa: E501
149
150 @classmethod
151 def create(
152 cls,
153 drivername: str,
154 username: Optional[str] = None,
155 password: Optional[str] = None,
156 host: Optional[str] = None,
157 port: Optional[int] = None,
158 database: Optional[str] = None,
159 query: Mapping[str, Union[Sequence[str], str]] = util.EMPTY_DICT,
160 ) -> URL:
161 """Create a new :class:`_engine.URL` object.
162
163 .. seealso::
164
165 :ref:`database_urls`
166
167 :param drivername: the name of the database backend. This name will
168 correspond to a module in sqlalchemy/databases or a third party
169 plug-in.
170 :param username: The user name.
171 :param password: database password. Is typically a string, but may
172 also be an object that can be stringified with ``str()``.
173
174 .. note:: The password string should **not** be URL encoded when
175 passed as an argument to :meth:`_engine.URL.create`; the string
176 should contain the password characters exactly as they would be
177 typed.
178
179 .. note:: A password-producing object will be stringified only
180 **once** per :class:`_engine.Engine` object. For dynamic password
181 generation per connect, see :ref:`engines_dynamic_tokens`.
182
183 :param host: The name of the host.
184 :param port: The port number.
185 :param database: The database name.
186 :param query: A dictionary of string keys to string values to be passed
187 to the dialect and/or the DBAPI upon connect. To specify non-string
188 parameters to a Python DBAPI directly, use the
189 :paramref:`_sa.create_engine.connect_args` parameter to
190 :func:`_sa.create_engine`. See also
191 :attr:`_engine.URL.normalized_query` for a dictionary that is
192 consistently string->list of string.
193 :return: new :class:`_engine.URL` object.
194
195 .. versionadded:: 1.4
196
197 The :class:`_engine.URL` object is now an **immutable named
198 tuple**. In addition, the ``query`` dictionary is also immutable.
199 To create a URL, use the :func:`_engine.url.make_url` or
200 :meth:`_engine.URL.create` function/ method. To modify a
201 :class:`_engine.URL`, use the :meth:`_engine.URL.set` and
202 :meth:`_engine.URL.update_query` methods.
203
204 """
205
206 return cls(
207 cls._assert_str(drivername, "drivername"),
208 cls._assert_none_str(username, "username"),
209 password,
210 cls._assert_none_str(host, "host"),
211 cls._assert_port(port),
212 cls._assert_none_str(database, "database"),
213 cls._str_dict(query),
214 )
215
216 @classmethod
217 def _assert_port(cls, port: Optional[int]) -> Optional[int]:
218 if port is None:
219 return None
220 try:
221 return int(port)
222 except TypeError:
223 raise TypeError("Port argument must be an integer or None")
224
225 @classmethod
226 def _assert_str(cls, v: str, paramname: str) -> str:
227 if not isinstance(v, str):
228 raise TypeError("%s must be a string" % paramname)
229 return v
230
231 @classmethod
232 def _assert_none_str(
233 cls, v: Optional[str], paramname: str
234 ) -> Optional[str]:
235 if v is None:
236 return v
237
238 return cls._assert_str(v, paramname)
239
240 @classmethod
241 def _str_dict(
242 cls,
243 dict_: Optional[
244 Union[
245 Sequence[Tuple[str, Union[Sequence[str], str]]],
246 Mapping[str, Union[Sequence[str], str]],
247 ]
248 ],
249 ) -> util.immutabledict[str, Union[Tuple[str, ...], str]]:
250 if dict_ is None:
251 return util.EMPTY_DICT
252
253 @overload
254 def _assert_value(
255 val: str,
256 ) -> str: ...
257
258 @overload
259 def _assert_value(
260 val: Sequence[str],
261 ) -> Union[str, Tuple[str, ...]]: ...
262
263 def _assert_value(
264 val: Union[str, Sequence[str]],
265 ) -> Union[str, Tuple[str, ...]]:
266 if isinstance(val, str):
267 return val
268 elif isinstance(val, collections_abc.Sequence):
269 return tuple(_assert_value(elem) for elem in val)
270 else:
271 raise TypeError(
272 "Query dictionary values must be strings or "
273 "sequences of strings"
274 )
275
276 def _assert_str(v: str) -> str:
277 if not isinstance(v, str):
278 raise TypeError("Query dictionary keys must be strings")
279 return v
280
281 dict_items: Iterable[Tuple[str, Union[Sequence[str], str]]]
282 if isinstance(dict_, collections_abc.Sequence):
283 dict_items = dict_
284 else:
285 dict_items = dict_.items()
286
287 return util.immutabledict(
288 {
289 _assert_str(key): _assert_value(
290 value,
291 )
292 for key, value in dict_items
293 }
294 )
295
296 def set(
297 self,
298 drivername: Optional[str] = None,
299 username: Optional[str] = None,
300 password: Optional[str] = None,
301 host: Optional[str] = None,
302 port: Optional[int] = None,
303 database: Optional[str] = None,
304 query: Optional[Mapping[str, Union[Sequence[str], str]]] = None,
305 ) -> URL:
306 """return a new :class:`_engine.URL` object with modifications.
307
308 Values are used if they are non-None. To set a value to ``None``
309 explicitly, use the :meth:`_engine.URL._replace` method adapted
310 from ``namedtuple``.
311
312 :param drivername: new drivername
313 :param username: new username
314 :param password: new password
315 :param host: new hostname
316 :param port: new port
317 :param query: new query parameters, passed a dict of string keys
318 referring to string or sequence of string values. Fully
319 replaces the previous list of arguments.
320
321 :return: new :class:`_engine.URL` object.
322
323 .. versionadded:: 1.4
324
325 .. seealso::
326
327 :meth:`_engine.URL.update_query_dict`
328
329 """
330
331 kw: Dict[str, Any] = {}
332 if drivername is not None:
333 kw["drivername"] = drivername
334 if username is not None:
335 kw["username"] = username
336 if password is not None:
337 kw["password"] = password
338 if host is not None:
339 kw["host"] = host
340 if port is not None:
341 kw["port"] = port
342 if database is not None:
343 kw["database"] = database
344 if query is not None:
345 kw["query"] = query
346
347 return self._assert_replace(**kw)
348
349 def _assert_replace(self, **kw: Any) -> URL:
350 """argument checks before calling _replace()"""
351
352 if "drivername" in kw:
353 self._assert_str(kw["drivername"], "drivername")
354 for name in "username", "host", "database":
355 if name in kw:
356 self._assert_none_str(kw[name], name)
357 if "port" in kw:
358 self._assert_port(kw["port"])
359 if "query" in kw:
360 kw["query"] = self._str_dict(kw["query"])
361
362 return self._replace(**kw)
363
364 def update_query_string(
365 self, query_string: str, append: bool = False
366 ) -> URL:
367 """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
368 parameter dictionary updated by the given query string.
369
370 E.g.::
371
372 >>> from sqlalchemy.engine import make_url
373 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
374 >>> url = url.update_query_string("alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
375 >>> str(url)
376 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
377
378 :param query_string: a URL escaped query string, not including the
379 question mark.
380
381 :param append: if True, parameters in the existing query string will
382 not be removed; new parameters will be in addition to those present.
383 If left at its default of False, keys present in the given query
384 parameters will replace those of the existing query string.
385
386 .. versionadded:: 1.4
387
388 .. seealso::
389
390 :attr:`_engine.URL.query`
391
392 :meth:`_engine.URL.update_query_dict`
393
394 """ # noqa: E501
395 return self.update_query_pairs(parse_qsl(query_string), append=append)
396
397 def update_query_pairs(
398 self,
399 key_value_pairs: Iterable[Tuple[str, Union[str, List[str]]]],
400 append: bool = False,
401 ) -> URL:
402 """Return a new :class:`_engine.URL` object with the
403 :attr:`_engine.URL.query`
404 parameter dictionary updated by the given sequence of key/value pairs
405
406 E.g.::
407
408 >>> from sqlalchemy.engine import make_url
409 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
410 >>> url = url.update_query_pairs([("alt_host", "host1"), ("alt_host", "host2"), ("ssl_cipher", "/path/to/crt")])
411 >>> str(url)
412 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
413
414 :param key_value_pairs: A sequence of tuples containing two strings
415 each.
416
417 :param append: if True, parameters in the existing query string will
418 not be removed; new parameters will be in addition to those present.
419 If left at its default of False, keys present in the given query
420 parameters will replace those of the existing query string.
421
422 .. versionadded:: 1.4
423
424 .. seealso::
425
426 :attr:`_engine.URL.query`
427
428 :meth:`_engine.URL.difference_update_query`
429
430 :meth:`_engine.URL.set`
431
432 """ # noqa: E501
433
434 existing_query = self.query
435 new_keys: Dict[str, Union[str, List[str]]] = {}
436
437 for key, value in key_value_pairs:
438 if key in new_keys:
439 new_keys[key] = util.to_list(new_keys[key])
440 cast("List[str]", new_keys[key]).append(cast(str, value))
441 else:
442 new_keys[key] = (
443 list(value) if isinstance(value, (list, tuple)) else value
444 )
445
446 new_query: Mapping[str, Union[str, Sequence[str]]]
447 if append:
448 new_query = {}
449
450 for k in new_keys:
451 if k in existing_query:
452 new_query[k] = tuple(
453 util.to_list(existing_query[k])
454 + util.to_list(new_keys[k])
455 )
456 else:
457 new_query[k] = new_keys[k]
458
459 new_query.update(
460 {
461 k: existing_query[k]
462 for k in set(existing_query).difference(new_keys)
463 }
464 )
465 else:
466 new_query = self.query.union(
467 {
468 k: tuple(v) if isinstance(v, list) else v
469 for k, v in new_keys.items()
470 }
471 )
472 return self.set(query=new_query)
473
474 def update_query_dict(
475 self,
476 query_parameters: Mapping[str, Union[str, List[str]]],
477 append: bool = False,
478 ) -> URL:
479 """Return a new :class:`_engine.URL` object with the
480 :attr:`_engine.URL.query` parameter dictionary updated by the given
481 dictionary.
482
483 The dictionary typically contains string keys and string values.
484 In order to represent a query parameter that is expressed multiple
485 times, pass a sequence of string values.
486
487 E.g.::
488
489
490 >>> from sqlalchemy.engine import make_url
491 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
492 >>> url = url.update_query_dict({"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"})
493 >>> str(url)
494 'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
495
496
497 :param query_parameters: A dictionary with string keys and values
498 that are either strings, or sequences of strings.
499
500 :param append: if True, parameters in the existing query string will
501 not be removed; new parameters will be in addition to those present.
502 If left at its default of False, keys present in the given query
503 parameters will replace those of the existing query string.
504
505
506 .. versionadded:: 1.4
507
508 .. seealso::
509
510 :attr:`_engine.URL.query`
511
512 :meth:`_engine.URL.update_query_string`
513
514 :meth:`_engine.URL.update_query_pairs`
515
516 :meth:`_engine.URL.difference_update_query`
517
518 :meth:`_engine.URL.set`
519
520 """ # noqa: E501
521 return self.update_query_pairs(query_parameters.items(), append=append)
522
523 def difference_update_query(self, names: Iterable[str]) -> URL:
524 """
525 Remove the given names from the :attr:`_engine.URL.query` dictionary,
526 returning the new :class:`_engine.URL`.
527
528 E.g.::
529
530 url = url.difference_update_query(['foo', 'bar'])
531
532 Equivalent to using :meth:`_engine.URL.set` as follows::
533
534 url = url.set(
535 query={
536 key: url.query[key]
537 for key in set(url.query).difference(['foo', 'bar'])
538 }
539 )
540
541 .. versionadded:: 1.4
542
543 .. seealso::
544
545 :attr:`_engine.URL.query`
546
547 :meth:`_engine.URL.update_query_dict`
548
549 :meth:`_engine.URL.set`
550
551 """
552
553 if not set(names).intersection(self.query):
554 return self
555
556 return URL(
557 self.drivername,
558 self.username,
559 self.password,
560 self.host,
561 self.port,
562 self.database,
563 util.immutabledict(
564 {
565 key: self.query[key]
566 for key in set(self.query).difference(names)
567 }
568 ),
569 )
570
571 @property
572 def normalized_query(self) -> Mapping[str, Sequence[str]]:
573 """Return the :attr:`_engine.URL.query` dictionary with values normalized
574 into sequences.
575
576 As the :attr:`_engine.URL.query` dictionary may contain either
577 string values or sequences of string values to differentiate between
578 parameters that are specified multiple times in the query string,
579 code that needs to handle multiple parameters generically will wish
580 to use this attribute so that all parameters present are presented
581 as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
582 function. E.g.::
583
584
585 >>> from sqlalchemy.engine import make_url
586 >>> url = make_url("postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt")
587 >>> url.query
588 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
589 >>> url.normalized_query
590 immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
591
592 """ # noqa: E501
593
594 return util.immutabledict(
595 {
596 k: (v,) if not isinstance(v, tuple) else v
597 for k, v in self.query.items()
598 }
599 )
600
601 @util.deprecated(
602 "1.4",
603 "The :meth:`_engine.URL.__to_string__ method is deprecated and will "
604 "be removed in a future release. Please use the "
605 ":meth:`_engine.URL.render_as_string` method.",
606 )
607 def __to_string__(self, hide_password: bool = True) -> str:
608 """Render this :class:`_engine.URL` object as a string.
609
610 :param hide_password: Defaults to True. The password is not shown
611 in the string unless this is set to False.
612
613 """
614 return self.render_as_string(hide_password=hide_password)
615
616 def render_as_string(self, hide_password: bool = True) -> str:
617 """Render this :class:`_engine.URL` object as a string.
618
619 This method is used when the ``__str__()`` or ``__repr__()``
620 methods are used. The method directly includes additional options.
621
622 :param hide_password: Defaults to True. The password is not shown
623 in the string unless this is set to False.
624
625 """
626 s = self.drivername + "://"
627 if self.username is not None:
628 s += quote(self.username, safe=" +")
629 if self.password is not None:
630 s += ":" + (
631 "***"
632 if hide_password
633 else quote(str(self.password), safe=" +")
634 )
635 s += "@"
636 if self.host is not None:
637 if ":" in self.host:
638 s += f"[{self.host}]"
639 else:
640 s += self.host
641 if self.port is not None:
642 s += ":" + str(self.port)
643 if self.database is not None:
644 s += "/" + self.database
645 if self.query:
646 keys = list(self.query)
647 keys.sort()
648 s += "?" + "&".join(
649 f"{quote_plus(k)}={quote_plus(element)}"
650 for k in keys
651 for element in util.to_list(self.query[k])
652 )
653 return s
654
655 def __repr__(self) -> str:
656 return self.render_as_string()
657
658 def __copy__(self) -> URL:
659 return self.__class__.create(
660 self.drivername,
661 self.username,
662 self.password,
663 self.host,
664 self.port,
665 self.database,
666 # note this is an immutabledict of str-> str / tuple of str,
667 # also fully immutable. does not require deepcopy
668 self.query,
669 )
670
671 def __deepcopy__(self, memo: Any) -> URL:
672 return self.__copy__()
673
674 def __hash__(self) -> int:
675 return hash(str(self))
676
677 def __eq__(self, other: Any) -> bool:
678 return (
679 isinstance(other, URL)
680 and self.drivername == other.drivername
681 and self.username == other.username
682 and self.password == other.password
683 and self.host == other.host
684 and self.database == other.database
685 and self.query == other.query
686 and self.port == other.port
687 )
688
689 def __ne__(self, other: Any) -> bool:
690 return not self == other
691
692 def get_backend_name(self) -> str:
693 """Return the backend name.
694
695 This is the name that corresponds to the database backend in
696 use, and is the portion of the :attr:`_engine.URL.drivername`
697 that is to the left of the plus sign.
698
699 """
700 if "+" not in self.drivername:
701 return self.drivername
702 else:
703 return self.drivername.split("+")[0]
704
705 def get_driver_name(self) -> str:
706 """Return the backend name.
707
708 This is the name that corresponds to the DBAPI driver in
709 use, and is the portion of the :attr:`_engine.URL.drivername`
710 that is to the right of the plus sign.
711
712 If the :attr:`_engine.URL.drivername` does not include a plus sign,
713 then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
714 is imported in order to get the driver name.
715
716 """
717
718 if "+" not in self.drivername:
719 return self.get_dialect().driver
720 else:
721 return self.drivername.split("+")[1]
722
723 def _instantiate_plugins(
724 self, kwargs: Mapping[str, Any]
725 ) -> Tuple[URL, List[Any], Dict[str, Any]]:
726 plugin_names = util.to_list(self.query.get("plugin", ()))
727 plugin_names += kwargs.get("plugins", [])
728
729 kwargs = dict(kwargs)
730
731 loaded_plugins = [
732 plugins.load(plugin_name)(self, kwargs)
733 for plugin_name in plugin_names
734 ]
735
736 u = self.difference_update_query(["plugin", "plugins"])
737
738 for plugin in loaded_plugins:
739 new_u = plugin.update_url(u)
740 if new_u is not None:
741 u = new_u
742
743 kwargs.pop("plugins", None)
744
745 return u, loaded_plugins, kwargs
746
747 def _get_entrypoint(self) -> Type[Dialect]:
748 """Return the "entry point" dialect class.
749
750 This is normally the dialect itself except in the case when the
751 returned class implements the get_dialect_cls() method.
752
753 """
754 if "+" not in self.drivername:
755 name = self.drivername
756 else:
757 name = self.drivername.replace("+", ".")
758 cls = registry.load(name)
759 # check for legacy dialects that
760 # would return a module with 'dialect' as the
761 # actual class
762 if (
763 hasattr(cls, "dialect")
764 and isinstance(cls.dialect, type)
765 and issubclass(cls.dialect, Dialect)
766 ):
767 return cls.dialect
768 else:
769 return cast("Type[Dialect]", cls)
770
771 def get_dialect(self, _is_async: bool = False) -> Type[Dialect]:
772 """Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
773 to this URL's driver name.
774
775 """
776 entrypoint = self._get_entrypoint()
777 if _is_async:
778 dialect_cls = entrypoint.get_async_dialect_cls(self)
779 else:
780 dialect_cls = entrypoint.get_dialect_cls(self)
781 return dialect_cls
782
783 def translate_connect_args(
784 self, names: Optional[List[str]] = None, **kw: Any
785 ) -> Dict[str, Any]:
786 r"""Translate url attributes into a dictionary of connection arguments.
787
788 Returns attributes of this url (`host`, `database`, `username`,
789 `password`, `port`) as a plain dictionary. The attribute names are
790 used as the keys by default. Unset or false attributes are omitted
791 from the final dictionary.
792
793 :param \**kw: Optional, alternate key names for url attributes.
794
795 :param names: Deprecated. Same purpose as the keyword-based alternate
796 names, but correlates the name to the original positionally.
797 """
798
799 if names is not None:
800 util.warn_deprecated(
801 "The `URL.translate_connect_args.name`s parameter is "
802 "deprecated. Please pass the "
803 "alternate names as kw arguments.",
804 "1.4",
805 )
806
807 translated = {}
808 attribute_names = ["host", "database", "username", "password", "port"]
809 for sname in attribute_names:
810 if names:
811 name = names.pop(0)
812 elif sname in kw:
813 name = kw[sname]
814 else:
815 name = sname
816 if name is not None and getattr(self, sname, False):
817 if sname == "password":
818 translated[name] = str(getattr(self, sname))
819 else:
820 translated[name] = getattr(self, sname)
821
822 return translated
823
824
825def make_url(name_or_url: Union[str, URL]) -> URL:
826 """Given a string, produce a new URL instance.
827
828 The format of the URL generally follows `RFC-1738
829 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
830 that underscores, and not dashes or periods, are accepted within the
831 "scheme" portion.
832
833 If a :class:`.URL` object is passed, it is returned as is.
834
835 .. seealso::
836
837 :ref:`database_urls`
838
839 """
840
841 if isinstance(name_or_url, str):
842 return _parse_url(name_or_url)
843 elif not isinstance(name_or_url, URL) and not hasattr(
844 name_or_url, "_sqla_is_testing_if_this_is_a_mock_object"
845 ):
846 raise exc.ArgumentError(
847 f"Expected string or URL object, got {name_or_url!r}"
848 )
849 else:
850 return name_or_url
851
852
853def _parse_url(name: str) -> URL:
854 pattern = re.compile(
855 r"""
856 (?P<name>[\w\+]+)://
857 (?:
858 (?P<username>[^:/]*)
859 (?::(?P<password>[^@]*))?
860 @)?
861 (?:
862 (?:
863 \[(?P<ipv6host>[^/\?]+)\] |
864 (?P<ipv4host>[^/:\?]+)
865 )?
866 (?::(?P<port>[^/\?]*))?
867 )?
868 (?:/(?P<database>[^\?]*))?
869 (?:\?(?P<query>.*))?
870 """,
871 re.X,
872 )
873
874 m = pattern.match(name)
875 if m is not None:
876 components = m.groupdict()
877 query: Optional[Dict[str, Union[str, List[str]]]]
878 if components["query"] is not None:
879 query = {}
880
881 for key, value in parse_qsl(components["query"]):
882 if key in query:
883 query[key] = util.to_list(query[key])
884 cast("List[str]", query[key]).append(value)
885 else:
886 query[key] = value
887 else:
888 query = None
889 components["query"] = query
890
891 if components["username"] is not None:
892 components["username"] = unquote(components["username"])
893
894 if components["password"] is not None:
895 components["password"] = unquote(components["password"])
896
897 ipv4host = components.pop("ipv4host")
898 ipv6host = components.pop("ipv6host")
899 components["host"] = ipv4host or ipv6host
900 name = components.pop("name")
901
902 if components["port"]:
903 components["port"] = int(components["port"])
904
905 return URL.create(name, **components) # type: ignore
906
907 else:
908 raise exc.ArgumentError(
909 "Could not parse SQLAlchemy URL from string '%s'" % name
910 )
911