Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/url.py: 44%

1# engine/url.py

3# <see AUTHORS file>

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``

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 )