Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/engine»
This commit is contained in:
400
venv/Lib/site-packages/sqlalchemy/engine/row.py
Normal file
400
venv/Lib/site-packages/sqlalchemy/engine/row.py
Normal file
@@ -0,0 +1,400 @@
|
||||
# engine/row.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
|
||||
"""Define row constructs including :class:`.Row`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC
|
||||
import collections.abc as collections_abc
|
||||
import operator
|
||||
import typing
|
||||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import Dict
|
||||
from typing import Generic
|
||||
from typing import Iterator
|
||||
from typing import List
|
||||
from typing import Mapping
|
||||
from typing import NoReturn
|
||||
from typing import Optional
|
||||
from typing import overload
|
||||
from typing import Sequence
|
||||
from typing import Tuple
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TypeVar
|
||||
from typing import Union
|
||||
|
||||
from ..sql import util as sql_util
|
||||
from ..util import deprecated
|
||||
from ..util._has_cy import HAS_CYEXTENSION
|
||||
|
||||
if TYPE_CHECKING or not HAS_CYEXTENSION:
|
||||
from ._py_row import BaseRow as BaseRow
|
||||
else:
|
||||
from sqlalchemy.cyextension.resultproxy import BaseRow as BaseRow
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .result import _KeyType
|
||||
from .result import _ProcessorsType
|
||||
from .result import RMKeyView
|
||||
|
||||
_T = TypeVar("_T", bound=Any)
|
||||
_TP = TypeVar("_TP", bound=Tuple[Any, ...])
|
||||
|
||||
|
||||
class Row(BaseRow, Sequence[Any], Generic[_TP]):
|
||||
"""Represent a single result row.
|
||||
|
||||
The :class:`.Row` object represents a row of a database result. It is
|
||||
typically associated in the 1.x series of SQLAlchemy with the
|
||||
:class:`_engine.CursorResult` object, however is also used by the ORM for
|
||||
tuple-like results as of SQLAlchemy 1.4.
|
||||
|
||||
The :class:`.Row` object seeks to act as much like a Python named
|
||||
tuple as possible. For mapping (i.e. dictionary) behavior on a row,
|
||||
such as testing for containment of keys, refer to the :attr:`.Row._mapping`
|
||||
attribute.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:ref:`tutorial_selecting_data` - includes examples of selecting
|
||||
rows from SELECT statements.
|
||||
|
||||
.. versionchanged:: 1.4
|
||||
|
||||
Renamed ``RowProxy`` to :class:`.Row`. :class:`.Row` is no longer a
|
||||
"proxy" object in that it contains the final form of data within it,
|
||||
and now acts mostly like a named tuple. Mapping-like functionality is
|
||||
moved to the :attr:`.Row._mapping` attribute. See
|
||||
:ref:`change_4710_core` for background on this change.
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> NoReturn:
|
||||
raise AttributeError("can't set attribute")
|
||||
|
||||
def __delattr__(self, name: str) -> NoReturn:
|
||||
raise AttributeError("can't delete attribute")
|
||||
|
||||
def _tuple(self) -> _TP:
|
||||
"""Return a 'tuple' form of this :class:`.Row`.
|
||||
|
||||
At runtime, this method returns "self"; the :class:`.Row` object is
|
||||
already a named tuple. However, at the typing level, if this
|
||||
:class:`.Row` is typed, the "tuple" return type will be a :pep:`484`
|
||||
``Tuple`` datatype that contains typing information about individual
|
||||
elements, supporting typed unpacking and attribute access.
|
||||
|
||||
.. versionadded:: 2.0.19 - The :meth:`.Row._tuple` method supersedes
|
||||
the previous :meth:`.Row.tuple` method, which is now underscored
|
||||
to avoid name conflicts with column names in the same way as other
|
||||
named-tuple methods on :class:`.Row`.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`.Row._t` - shorthand attribute notation
|
||||
|
||||
:meth:`.Result.tuples`
|
||||
|
||||
|
||||
"""
|
||||
return self # type: ignore
|
||||
|
||||
@deprecated(
|
||||
"2.0.19",
|
||||
"The :meth:`.Row.tuple` method is deprecated in favor of "
|
||||
":meth:`.Row._tuple`; all :class:`.Row` "
|
||||
"methods and library-level attributes are intended to be underscored "
|
||||
"to avoid name conflicts. Please use :meth:`Row._tuple`.",
|
||||
)
|
||||
def tuple(self) -> _TP:
|
||||
"""Return a 'tuple' form of this :class:`.Row`.
|
||||
|
||||
.. versionadded:: 2.0
|
||||
|
||||
"""
|
||||
return self._tuple()
|
||||
|
||||
@property
|
||||
def _t(self) -> _TP:
|
||||
"""A synonym for :meth:`.Row._tuple`.
|
||||
|
||||
.. versionadded:: 2.0.19 - The :attr:`.Row._t` attribute supersedes
|
||||
the previous :attr:`.Row.t` attribute, which is now underscored
|
||||
to avoid name conflicts with column names in the same way as other
|
||||
named-tuple methods on :class:`.Row`.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`.Result.t`
|
||||
"""
|
||||
return self # type: ignore
|
||||
|
||||
@property
|
||||
@deprecated(
|
||||
"2.0.19",
|
||||
"The :attr:`.Row.t` attribute is deprecated in favor of "
|
||||
":attr:`.Row._t`; all :class:`.Row` "
|
||||
"methods and library-level attributes are intended to be underscored "
|
||||
"to avoid name conflicts. Please use :attr:`Row._t`.",
|
||||
)
|
||||
def t(self) -> _TP:
|
||||
"""A synonym for :meth:`.Row._tuple`.
|
||||
|
||||
.. versionadded:: 2.0
|
||||
|
||||
"""
|
||||
return self._t
|
||||
|
||||
@property
|
||||
def _mapping(self) -> RowMapping:
|
||||
"""Return a :class:`.RowMapping` for this :class:`.Row`.
|
||||
|
||||
This object provides a consistent Python mapping (i.e. dictionary)
|
||||
interface for the data contained within the row. The :class:`.Row`
|
||||
by itself behaves like a named tuple.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`.Row._fields`
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
"""
|
||||
return RowMapping(self._parent, None, self._key_to_index, self._data)
|
||||
|
||||
def _filter_on_values(
|
||||
self, processor: Optional[_ProcessorsType]
|
||||
) -> Row[Any]:
|
||||
return Row(self._parent, processor, self._key_to_index, self._data)
|
||||
|
||||
if not TYPE_CHECKING:
|
||||
|
||||
def _special_name_accessor(name: str) -> Any:
|
||||
"""Handle ambiguous names such as "count" and "index" """
|
||||
|
||||
@property
|
||||
def go(self: Row) -> Any:
|
||||
if self._parent._has_key(name):
|
||||
return self.__getattr__(name)
|
||||
else:
|
||||
|
||||
def meth(*arg: Any, **kw: Any) -> Any:
|
||||
return getattr(collections_abc.Sequence, name)(
|
||||
self, *arg, **kw
|
||||
)
|
||||
|
||||
return meth
|
||||
|
||||
return go
|
||||
|
||||
count = _special_name_accessor("count")
|
||||
index = _special_name_accessor("index")
|
||||
|
||||
def __contains__(self, key: Any) -> bool:
|
||||
return key in self._data
|
||||
|
||||
def _op(self, other: Any, op: Callable[[Any, Any], bool]) -> bool:
|
||||
return (
|
||||
op(self._to_tuple_instance(), other._to_tuple_instance())
|
||||
if isinstance(other, Row)
|
||||
else op(self._to_tuple_instance(), other)
|
||||
)
|
||||
|
||||
__hash__ = BaseRow.__hash__
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
@overload
|
||||
def __getitem__(self, index: int) -> Any: ...
|
||||
|
||||
@overload
|
||||
def __getitem__(self, index: slice) -> Sequence[Any]: ...
|
||||
|
||||
def __getitem__(self, index: Union[int, slice]) -> Any: ...
|
||||
|
||||
def __lt__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.lt)
|
||||
|
||||
def __le__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.le)
|
||||
|
||||
def __ge__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.ge)
|
||||
|
||||
def __gt__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.gt)
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.eq)
|
||||
|
||||
def __ne__(self, other: Any) -> bool:
|
||||
return self._op(other, operator.ne)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return repr(sql_util._repr_row(self))
|
||||
|
||||
@property
|
||||
def _fields(self) -> Tuple[str, ...]:
|
||||
"""Return a tuple of string keys as represented by this
|
||||
:class:`.Row`.
|
||||
|
||||
The keys can represent the labels of the columns returned by a core
|
||||
statement or the names of the orm classes returned by an orm
|
||||
execution.
|
||||
|
||||
This attribute is analogous to the Python named tuple ``._fields``
|
||||
attribute.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`.Row._mapping`
|
||||
|
||||
"""
|
||||
return tuple([k for k in self._parent.keys if k is not None])
|
||||
|
||||
def _asdict(self) -> Dict[str, Any]:
|
||||
"""Return a new dict which maps field names to their corresponding
|
||||
values.
|
||||
|
||||
This method is analogous to the Python named tuple ``._asdict()``
|
||||
method, and works by applying the ``dict()`` constructor to the
|
||||
:attr:`.Row._mapping` attribute.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`.Row._mapping`
|
||||
|
||||
"""
|
||||
return dict(self._mapping)
|
||||
|
||||
|
||||
BaseRowProxy = BaseRow
|
||||
RowProxy = Row
|
||||
|
||||
|
||||
class ROMappingView(ABC):
|
||||
__slots__ = ()
|
||||
|
||||
_items: Sequence[Any]
|
||||
_mapping: Mapping["_KeyType", Any]
|
||||
|
||||
def __init__(
|
||||
self, mapping: Mapping["_KeyType", Any], items: Sequence[Any]
|
||||
):
|
||||
self._mapping = mapping # type: ignore[misc]
|
||||
self._items = items # type: ignore[misc]
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._items)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "{0.__class__.__name__}({0._mapping!r})".format(self)
|
||||
|
||||
def __iter__(self) -> Iterator[Any]:
|
||||
return iter(self._items)
|
||||
|
||||
def __contains__(self, item: Any) -> bool:
|
||||
return item in self._items
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
return list(other) == list(self)
|
||||
|
||||
def __ne__(self, other: Any) -> bool:
|
||||
return list(other) != list(self)
|
||||
|
||||
|
||||
class ROMappingKeysValuesView(
|
||||
ROMappingView, typing.KeysView["_KeyType"], typing.ValuesView[Any]
|
||||
):
|
||||
__slots__ = ("_items",) # mapping slot is provided by KeysView
|
||||
|
||||
|
||||
class ROMappingItemsView(ROMappingView, typing.ItemsView["_KeyType", Any]):
|
||||
__slots__ = ("_items",) # mapping slot is provided by ItemsView
|
||||
|
||||
|
||||
class RowMapping(BaseRow, typing.Mapping["_KeyType", Any]):
|
||||
"""A ``Mapping`` that maps column names and objects to :class:`.Row`
|
||||
values.
|
||||
|
||||
The :class:`.RowMapping` is available from a :class:`.Row` via the
|
||||
:attr:`.Row._mapping` attribute, as well as from the iterable interface
|
||||
provided by the :class:`.MappingResult` object returned by the
|
||||
:meth:`_engine.Result.mappings` method.
|
||||
|
||||
:class:`.RowMapping` supplies Python mapping (i.e. dictionary) access to
|
||||
the contents of the row. This includes support for testing of
|
||||
containment of specific keys (string column names or objects), as well
|
||||
as iteration of keys, values, and items::
|
||||
|
||||
for row in result:
|
||||
if "a" in row._mapping:
|
||||
print("Column 'a': %s" % row._mapping["a"])
|
||||
|
||||
print("Column b: %s" % row._mapping[table.c.b])
|
||||
|
||||
.. versionadded:: 1.4 The :class:`.RowMapping` object replaces the
|
||||
mapping-like access previously provided by a database result row,
|
||||
which now seeks to behave mostly like a named tuple.
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
def __getitem__(self, key: _KeyType) -> Any: ...
|
||||
|
||||
else:
|
||||
__getitem__ = BaseRow._get_by_key_impl_mapping
|
||||
|
||||
def _values_impl(self) -> List[Any]:
|
||||
return list(self._data)
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
return (k for k in self._parent.keys if k is not None)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._data)
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
return self._parent._has_key(key)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return repr(dict(self))
|
||||
|
||||
def items(self) -> ROMappingItemsView:
|
||||
"""Return a view of key/value tuples for the elements in the
|
||||
underlying :class:`.Row`.
|
||||
|
||||
"""
|
||||
return ROMappingItemsView(
|
||||
self, [(key, self[key]) for key in self.keys()]
|
||||
)
|
||||
|
||||
def keys(self) -> RMKeyView:
|
||||
"""Return a view of 'keys' for string column names represented
|
||||
by the underlying :class:`.Row`.
|
||||
|
||||
"""
|
||||
|
||||
return self._parent.keys
|
||||
|
||||
def values(self) -> ROMappingKeysValuesView:
|
||||
"""Return a view of values for the values represented in the
|
||||
underlying :class:`.Row`.
|
||||
|
||||
"""
|
||||
return ROMappingKeysValuesView(self, self._values_impl())
|
||||
16
venv/Lib/site-packages/sqlalchemy/engine/strategies.py
Normal file
16
venv/Lib/site-packages/sqlalchemy/engine/strategies.py
Normal file
@@ -0,0 +1,16 @@
|
||||
# engine/strategies.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
|
||||
"""Deprecated mock engine strategy used by Alembic."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .mock import MockConnection # noqa
|
||||
|
||||
|
||||
class MockEngineStrategy:
|
||||
MockConnection = MockConnection
|
||||
924
venv/Lib/site-packages/sqlalchemy/engine/url.py
Normal file
924
venv/Lib/site-packages/sqlalchemy/engine/url.py
Normal file
@@ -0,0 +1,924 @@
|
||||
# engine/url.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
|
||||
"""Provides the :class:`~sqlalchemy.engine.url.URL` class which encapsulates
|
||||
information about a database connection specification.
|
||||
|
||||
The URL object is created automatically when
|
||||
:func:`~sqlalchemy.engine.create_engine` is called with a string
|
||||
argument; alternatively, the URL is a public-facing construct which can
|
||||
be used directly and is also accepted directly by ``create_engine()``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import collections.abc as collections_abc
|
||||
import re
|
||||
from typing import Any
|
||||
from typing import cast
|
||||
from typing import Dict
|
||||
from typing import Iterable
|
||||
from typing import List
|
||||
from typing import Mapping
|
||||
from typing import NamedTuple
|
||||
from typing import Optional
|
||||
from typing import overload
|
||||
from typing import Sequence
|
||||
from typing import Tuple
|
||||
from typing import Type
|
||||
from typing import Union
|
||||
from urllib.parse import parse_qsl
|
||||
from urllib.parse import quote
|
||||
from urllib.parse import quote_plus
|
||||
from urllib.parse import unquote
|
||||
|
||||
from .interfaces import Dialect
|
||||
from .. import exc
|
||||
from .. import util
|
||||
from ..dialects import plugins
|
||||
from ..dialects import registry
|
||||
|
||||
|
||||
class URL(NamedTuple):
|
||||
"""
|
||||
Represent the components of a URL used to connect to a database.
|
||||
|
||||
URLs are typically constructed from a fully formatted URL string, where the
|
||||
:func:`.make_url` function is used internally by the
|
||||
:func:`_sa.create_engine` function in order to parse the URL string into
|
||||
its individual components, which are then used to construct a new
|
||||
:class:`.URL` object. When parsing from a formatted URL string, the parsing
|
||||
format generally follows
|
||||
`RFC-1738 <https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions.
|
||||
|
||||
A :class:`_engine.URL` object may also be produced directly, either by
|
||||
using the :func:`.make_url` function with a fully formed URL string, or
|
||||
by using the :meth:`_engine.URL.create` constructor in order
|
||||
to construct a :class:`_engine.URL` programmatically given individual
|
||||
fields. The resulting :class:`.URL` object may be passed directly to
|
||||
:func:`_sa.create_engine` in place of a string argument, which will bypass
|
||||
the usage of :func:`.make_url` within the engine's creation process.
|
||||
|
||||
.. versionchanged:: 1.4
|
||||
|
||||
The :class:`_engine.URL` object is now an immutable object. To
|
||||
create a URL, use the :func:`_engine.make_url` or
|
||||
:meth:`_engine.URL.create` function / method. To modify
|
||||
a :class:`_engine.URL`, use methods like
|
||||
:meth:`_engine.URL.set` and
|
||||
:meth:`_engine.URL.update_query_dict` to return a new
|
||||
:class:`_engine.URL` object with modifications. See notes for this
|
||||
change at :ref:`change_5526`.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:ref:`database_urls`
|
||||
|
||||
:class:`_engine.URL` contains the following attributes:
|
||||
|
||||
* :attr:`_engine.URL.drivername`: database backend and driver name, such as
|
||||
``postgresql+psycopg2``
|
||||
* :attr:`_engine.URL.username`: username string
|
||||
* :attr:`_engine.URL.password`: password string
|
||||
* :attr:`_engine.URL.host`: string hostname
|
||||
* :attr:`_engine.URL.port`: integer port number
|
||||
* :attr:`_engine.URL.database`: string database name
|
||||
* :attr:`_engine.URL.query`: an immutable mapping representing the query
|
||||
string. contains strings for keys and either strings or tuples of
|
||||
strings for values.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
drivername: str
|
||||
"""database backend and driver name, such as
|
||||
``postgresql+psycopg2``
|
||||
|
||||
"""
|
||||
|
||||
username: Optional[str]
|
||||
"username string"
|
||||
|
||||
password: Optional[str]
|
||||
"""password, which is normally a string but may also be any
|
||||
object that has a ``__str__()`` method."""
|
||||
|
||||
host: Optional[str]
|
||||
"""hostname or IP number. May also be a data source name for some
|
||||
drivers."""
|
||||
|
||||
port: Optional[int]
|
||||
"""integer port number"""
|
||||
|
||||
database: Optional[str]
|
||||
"""database name"""
|
||||
|
||||
query: util.immutabledict[str, Union[Tuple[str, ...], str]]
|
||||
"""an immutable mapping representing the query string. contains strings
|
||||
for keys and either strings or tuples of strings for values, e.g.::
|
||||
|
||||
>>> from sqlalchemy.engine import make_url
|
||||
>>> url = make_url(
|
||||
... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
|
||||
... )
|
||||
>>> url.query
|
||||
immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
|
||||
|
||||
To create a mutable copy of this mapping, use the ``dict`` constructor::
|
||||
|
||||
mutable_query_opts = dict(url.query)
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`_engine.URL.normalized_query` - normalizes all values into sequences
|
||||
for consistent processing
|
||||
|
||||
Methods for altering the contents of :attr:`_engine.URL.query`:
|
||||
|
||||
:meth:`_engine.URL.update_query_dict`
|
||||
|
||||
:meth:`_engine.URL.update_query_string`
|
||||
|
||||
:meth:`_engine.URL.update_query_pairs`
|
||||
|
||||
:meth:`_engine.URL.difference_update_query`
|
||||
|
||||
""" # noqa: E501
|
||||
|
||||
@classmethod
|
||||
def create(
|
||||
cls,
|
||||
drivername: str,
|
||||
username: Optional[str] = None,
|
||||
password: Optional[str] = None,
|
||||
host: Optional[str] = None,
|
||||
port: Optional[int] = None,
|
||||
database: Optional[str] = None,
|
||||
query: Mapping[str, Union[Sequence[str], str]] = util.EMPTY_DICT,
|
||||
) -> URL:
|
||||
"""Create a new :class:`_engine.URL` object.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:ref:`database_urls`
|
||||
|
||||
:param drivername: the name of the database backend. This name will
|
||||
correspond to a module in sqlalchemy/databases or a third party
|
||||
plug-in.
|
||||
:param username: The user name.
|
||||
:param password: database password. Is typically a string, but may
|
||||
also be an object that can be stringified with ``str()``.
|
||||
|
||||
.. note:: The password string should **not** be URL encoded when
|
||||
passed as an argument to :meth:`_engine.URL.create`; the string
|
||||
should contain the password characters exactly as they would be
|
||||
typed.
|
||||
|
||||
.. note:: A password-producing object will be stringified only
|
||||
**once** per :class:`_engine.Engine` object. For dynamic password
|
||||
generation per connect, see :ref:`engines_dynamic_tokens`.
|
||||
|
||||
:param host: The name of the host.
|
||||
:param port: The port number.
|
||||
:param database: The database name.
|
||||
:param query: A dictionary of string keys to string values to be passed
|
||||
to the dialect and/or the DBAPI upon connect. To specify non-string
|
||||
parameters to a Python DBAPI directly, use the
|
||||
:paramref:`_sa.create_engine.connect_args` parameter to
|
||||
:func:`_sa.create_engine`. See also
|
||||
:attr:`_engine.URL.normalized_query` for a dictionary that is
|
||||
consistently string->list of string.
|
||||
:return: new :class:`_engine.URL` object.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
The :class:`_engine.URL` object is now an **immutable named
|
||||
tuple**. In addition, the ``query`` dictionary is also immutable.
|
||||
To create a URL, use the :func:`_engine.url.make_url` or
|
||||
:meth:`_engine.URL.create` function/ method. To modify a
|
||||
:class:`_engine.URL`, use the :meth:`_engine.URL.set` and
|
||||
:meth:`_engine.URL.update_query` methods.
|
||||
|
||||
"""
|
||||
|
||||
return cls(
|
||||
cls._assert_str(drivername, "drivername"),
|
||||
cls._assert_none_str(username, "username"),
|
||||
password,
|
||||
cls._assert_none_str(host, "host"),
|
||||
cls._assert_port(port),
|
||||
cls._assert_none_str(database, "database"),
|
||||
cls._str_dict(query),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _assert_port(cls, port: Optional[int]) -> Optional[int]:
|
||||
if port is None:
|
||||
return None
|
||||
try:
|
||||
return int(port)
|
||||
except TypeError:
|
||||
raise TypeError("Port argument must be an integer or None")
|
||||
|
||||
@classmethod
|
||||
def _assert_str(cls, v: str, paramname: str) -> str:
|
||||
if not isinstance(v, str):
|
||||
raise TypeError("%s must be a string" % paramname)
|
||||
return v
|
||||
|
||||
@classmethod
|
||||
def _assert_none_str(
|
||||
cls, v: Optional[str], paramname: str
|
||||
) -> Optional[str]:
|
||||
if v is None:
|
||||
return v
|
||||
|
||||
return cls._assert_str(v, paramname)
|
||||
|
||||
@classmethod
|
||||
def _str_dict(
|
||||
cls,
|
||||
dict_: Optional[
|
||||
Union[
|
||||
Sequence[Tuple[str, Union[Sequence[str], str]]],
|
||||
Mapping[str, Union[Sequence[str], str]],
|
||||
]
|
||||
],
|
||||
) -> util.immutabledict[str, Union[Tuple[str, ...], str]]:
|
||||
if dict_ is None:
|
||||
return util.EMPTY_DICT
|
||||
|
||||
@overload
|
||||
def _assert_value(
|
||||
val: str,
|
||||
) -> str: ...
|
||||
|
||||
@overload
|
||||
def _assert_value(
|
||||
val: Sequence[str],
|
||||
) -> Union[str, Tuple[str, ...]]: ...
|
||||
|
||||
def _assert_value(
|
||||
val: Union[str, Sequence[str]],
|
||||
) -> Union[str, Tuple[str, ...]]:
|
||||
if isinstance(val, str):
|
||||
return val
|
||||
elif isinstance(val, collections_abc.Sequence):
|
||||
return tuple(_assert_value(elem) for elem in val)
|
||||
else:
|
||||
raise TypeError(
|
||||
"Query dictionary values must be strings or "
|
||||
"sequences of strings"
|
||||
)
|
||||
|
||||
def _assert_str(v: str) -> str:
|
||||
if not isinstance(v, str):
|
||||
raise TypeError("Query dictionary keys must be strings")
|
||||
return v
|
||||
|
||||
dict_items: Iterable[Tuple[str, Union[Sequence[str], str]]]
|
||||
if isinstance(dict_, collections_abc.Sequence):
|
||||
dict_items = dict_
|
||||
else:
|
||||
dict_items = dict_.items()
|
||||
|
||||
return util.immutabledict(
|
||||
{
|
||||
_assert_str(key): _assert_value(
|
||||
value,
|
||||
)
|
||||
for key, value in dict_items
|
||||
}
|
||||
)
|
||||
|
||||
def set(
|
||||
self,
|
||||
drivername: Optional[str] = None,
|
||||
username: Optional[str] = None,
|
||||
password: Optional[str] = None,
|
||||
host: Optional[str] = None,
|
||||
port: Optional[int] = None,
|
||||
database: Optional[str] = None,
|
||||
query: Optional[Mapping[str, Union[Sequence[str], str]]] = None,
|
||||
) -> URL:
|
||||
"""return a new :class:`_engine.URL` object with modifications.
|
||||
|
||||
Values are used if they are non-None. To set a value to ``None``
|
||||
explicitly, use the :meth:`_engine.URL._replace` method adapted
|
||||
from ``namedtuple``.
|
||||
|
||||
:param drivername: new drivername
|
||||
:param username: new username
|
||||
:param password: new password
|
||||
:param host: new hostname
|
||||
:param port: new port
|
||||
:param query: new query parameters, passed a dict of string keys
|
||||
referring to string or sequence of string values. Fully
|
||||
replaces the previous list of arguments.
|
||||
|
||||
:return: new :class:`_engine.URL` object.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:meth:`_engine.URL.update_query_dict`
|
||||
|
||||
"""
|
||||
|
||||
kw: Dict[str, Any] = {}
|
||||
if drivername is not None:
|
||||
kw["drivername"] = drivername
|
||||
if username is not None:
|
||||
kw["username"] = username
|
||||
if password is not None:
|
||||
kw["password"] = password
|
||||
if host is not None:
|
||||
kw["host"] = host
|
||||
if port is not None:
|
||||
kw["port"] = port
|
||||
if database is not None:
|
||||
kw["database"] = database
|
||||
if query is not None:
|
||||
kw["query"] = query
|
||||
|
||||
return self._assert_replace(**kw)
|
||||
|
||||
def _assert_replace(self, **kw: Any) -> URL:
|
||||
"""argument checks before calling _replace()"""
|
||||
|
||||
if "drivername" in kw:
|
||||
self._assert_str(kw["drivername"], "drivername")
|
||||
for name in "username", "host", "database":
|
||||
if name in kw:
|
||||
self._assert_none_str(kw[name], name)
|
||||
if "port" in kw:
|
||||
self._assert_port(kw["port"])
|
||||
if "query" in kw:
|
||||
kw["query"] = self._str_dict(kw["query"])
|
||||
|
||||
return self._replace(**kw)
|
||||
|
||||
def update_query_string(
|
||||
self, query_string: str, append: bool = False
|
||||
) -> URL:
|
||||
"""Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
|
||||
parameter dictionary updated by the given query string.
|
||||
|
||||
E.g.::
|
||||
|
||||
>>> from sqlalchemy.engine import make_url
|
||||
>>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
|
||||
>>> url = url.update_query_string(
|
||||
... "alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
|
||||
... )
|
||||
>>> str(url)
|
||||
'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
|
||||
|
||||
:param query_string: a URL escaped query string, not including the
|
||||
question mark.
|
||||
|
||||
:param append: if True, parameters in the existing query string will
|
||||
not be removed; new parameters will be in addition to those present.
|
||||
If left at its default of False, keys present in the given query
|
||||
parameters will replace those of the existing query string.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`_engine.URL.query`
|
||||
|
||||
:meth:`_engine.URL.update_query_dict`
|
||||
|
||||
""" # noqa: E501
|
||||
return self.update_query_pairs(parse_qsl(query_string), append=append)
|
||||
|
||||
def update_query_pairs(
|
||||
self,
|
||||
key_value_pairs: Iterable[Tuple[str, Union[str, List[str]]]],
|
||||
append: bool = False,
|
||||
) -> URL:
|
||||
"""Return a new :class:`_engine.URL` object with the
|
||||
:attr:`_engine.URL.query`
|
||||
parameter dictionary updated by the given sequence of key/value pairs
|
||||
|
||||
E.g.::
|
||||
|
||||
>>> from sqlalchemy.engine import make_url
|
||||
>>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
|
||||
>>> url = url.update_query_pairs(
|
||||
... [
|
||||
... ("alt_host", "host1"),
|
||||
... ("alt_host", "host2"),
|
||||
... ("ssl_cipher", "/path/to/crt"),
|
||||
... ]
|
||||
... )
|
||||
>>> str(url)
|
||||
'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
|
||||
|
||||
:param key_value_pairs: A sequence of tuples containing two strings
|
||||
each.
|
||||
|
||||
:param append: if True, parameters in the existing query string will
|
||||
not be removed; new parameters will be in addition to those present.
|
||||
If left at its default of False, keys present in the given query
|
||||
parameters will replace those of the existing query string.
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`_engine.URL.query`
|
||||
|
||||
:meth:`_engine.URL.difference_update_query`
|
||||
|
||||
:meth:`_engine.URL.set`
|
||||
|
||||
""" # noqa: E501
|
||||
|
||||
existing_query = self.query
|
||||
new_keys: Dict[str, Union[str, List[str]]] = {}
|
||||
|
||||
for key, value in key_value_pairs:
|
||||
if key in new_keys:
|
||||
new_keys[key] = util.to_list(new_keys[key])
|
||||
cast("List[str]", new_keys[key]).append(cast(str, value))
|
||||
else:
|
||||
new_keys[key] = (
|
||||
list(value) if isinstance(value, (list, tuple)) else value
|
||||
)
|
||||
|
||||
new_query: Mapping[str, Union[str, Sequence[str]]]
|
||||
if append:
|
||||
new_query = {}
|
||||
|
||||
for k in new_keys:
|
||||
if k in existing_query:
|
||||
new_query[k] = tuple(
|
||||
util.to_list(existing_query[k])
|
||||
+ util.to_list(new_keys[k])
|
||||
)
|
||||
else:
|
||||
new_query[k] = new_keys[k]
|
||||
|
||||
new_query.update(
|
||||
{
|
||||
k: existing_query[k]
|
||||
for k in set(existing_query).difference(new_keys)
|
||||
}
|
||||
)
|
||||
else:
|
||||
new_query = self.query.union(
|
||||
{
|
||||
k: tuple(v) if isinstance(v, list) else v
|
||||
for k, v in new_keys.items()
|
||||
}
|
||||
)
|
||||
return self.set(query=new_query)
|
||||
|
||||
def update_query_dict(
|
||||
self,
|
||||
query_parameters: Mapping[str, Union[str, List[str]]],
|
||||
append: bool = False,
|
||||
) -> URL:
|
||||
"""Return a new :class:`_engine.URL` object with the
|
||||
:attr:`_engine.URL.query` parameter dictionary updated by the given
|
||||
dictionary.
|
||||
|
||||
The dictionary typically contains string keys and string values.
|
||||
In order to represent a query parameter that is expressed multiple
|
||||
times, pass a sequence of string values.
|
||||
|
||||
E.g.::
|
||||
|
||||
|
||||
>>> from sqlalchemy.engine import make_url
|
||||
>>> url = make_url("postgresql+psycopg2://user:pass@host/dbname")
|
||||
>>> url = url.update_query_dict(
|
||||
... {"alt_host": ["host1", "host2"], "ssl_cipher": "/path/to/crt"}
|
||||
... )
|
||||
>>> str(url)
|
||||
'postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt'
|
||||
|
||||
|
||||
:param query_parameters: A dictionary with string keys and values
|
||||
that are either strings, or sequences of strings.
|
||||
|
||||
:param append: if True, parameters in the existing query string will
|
||||
not be removed; new parameters will be in addition to those present.
|
||||
If left at its default of False, keys present in the given query
|
||||
parameters will replace those of the existing query string.
|
||||
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`_engine.URL.query`
|
||||
|
||||
:meth:`_engine.URL.update_query_string`
|
||||
|
||||
:meth:`_engine.URL.update_query_pairs`
|
||||
|
||||
:meth:`_engine.URL.difference_update_query`
|
||||
|
||||
:meth:`_engine.URL.set`
|
||||
|
||||
""" # noqa: E501
|
||||
return self.update_query_pairs(query_parameters.items(), append=append)
|
||||
|
||||
def difference_update_query(self, names: Iterable[str]) -> URL:
|
||||
"""
|
||||
Remove the given names from the :attr:`_engine.URL.query` dictionary,
|
||||
returning the new :class:`_engine.URL`.
|
||||
|
||||
E.g.::
|
||||
|
||||
url = url.difference_update_query(["foo", "bar"])
|
||||
|
||||
Equivalent to using :meth:`_engine.URL.set` as follows::
|
||||
|
||||
url = url.set(
|
||||
query={
|
||||
key: url.query[key]
|
||||
for key in set(url.query).difference(["foo", "bar"])
|
||||
}
|
||||
)
|
||||
|
||||
.. versionadded:: 1.4
|
||||
|
||||
.. seealso::
|
||||
|
||||
:attr:`_engine.URL.query`
|
||||
|
||||
:meth:`_engine.URL.update_query_dict`
|
||||
|
||||
:meth:`_engine.URL.set`
|
||||
|
||||
"""
|
||||
|
||||
if not set(names).intersection(self.query):
|
||||
return self
|
||||
|
||||
return URL(
|
||||
self.drivername,
|
||||
self.username,
|
||||
self.password,
|
||||
self.host,
|
||||
self.port,
|
||||
self.database,
|
||||
util.immutabledict(
|
||||
{
|
||||
key: self.query[key]
|
||||
for key in set(self.query).difference(names)
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
@property
|
||||
def normalized_query(self) -> Mapping[str, Sequence[str]]:
|
||||
"""Return the :attr:`_engine.URL.query` dictionary with values normalized
|
||||
into sequences.
|
||||
|
||||
As the :attr:`_engine.URL.query` dictionary may contain either
|
||||
string values or sequences of string values to differentiate between
|
||||
parameters that are specified multiple times in the query string,
|
||||
code that needs to handle multiple parameters generically will wish
|
||||
to use this attribute so that all parameters present are presented
|
||||
as sequences. Inspiration is from Python's ``urllib.parse.parse_qs``
|
||||
function. E.g.::
|
||||
|
||||
|
||||
>>> from sqlalchemy.engine import make_url
|
||||
>>> url = make_url(
|
||||
... "postgresql+psycopg2://user:pass@host/dbname?alt_host=host1&alt_host=host2&ssl_cipher=%2Fpath%2Fto%2Fcrt"
|
||||
... )
|
||||
>>> url.query
|
||||
immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': '/path/to/crt'})
|
||||
>>> url.normalized_query
|
||||
immutabledict({'alt_host': ('host1', 'host2'), 'ssl_cipher': ('/path/to/crt',)})
|
||||
|
||||
""" # noqa: E501
|
||||
|
||||
return util.immutabledict(
|
||||
{
|
||||
k: (v,) if not isinstance(v, tuple) else v
|
||||
for k, v in self.query.items()
|
||||
}
|
||||
)
|
||||
|
||||
@util.deprecated(
|
||||
"1.4",
|
||||
"The :meth:`_engine.URL.__to_string__ method is deprecated and will "
|
||||
"be removed in a future release. Please use the "
|
||||
":meth:`_engine.URL.render_as_string` method.",
|
||||
)
|
||||
def __to_string__(self, hide_password: bool = True) -> str:
|
||||
"""Render this :class:`_engine.URL` object as a string.
|
||||
|
||||
:param hide_password: Defaults to True. The password is not shown
|
||||
in the string unless this is set to False.
|
||||
|
||||
"""
|
||||
return self.render_as_string(hide_password=hide_password)
|
||||
|
||||
def render_as_string(self, hide_password: bool = True) -> str:
|
||||
"""Render this :class:`_engine.URL` object as a string.
|
||||
|
||||
This method is used when the ``__str__()`` or ``__repr__()``
|
||||
methods are used. The method directly includes additional options.
|
||||
|
||||
:param hide_password: Defaults to True. The password is not shown
|
||||
in the string unless this is set to False.
|
||||
|
||||
"""
|
||||
s = self.drivername + "://"
|
||||
if self.username is not None:
|
||||
s += quote(self.username, safe=" +")
|
||||
if self.password is not None:
|
||||
s += ":" + (
|
||||
"***"
|
||||
if hide_password
|
||||
else quote(str(self.password), safe=" +")
|
||||
)
|
||||
s += "@"
|
||||
if self.host is not None:
|
||||
if ":" in self.host:
|
||||
s += f"[{self.host}]"
|
||||
else:
|
||||
s += self.host
|
||||
if self.port is not None:
|
||||
s += ":" + str(self.port)
|
||||
if self.database is not None:
|
||||
s += "/" + self.database
|
||||
if self.query:
|
||||
keys = list(self.query)
|
||||
keys.sort()
|
||||
s += "?" + "&".join(
|
||||
f"{quote_plus(k)}={quote_plus(element)}"
|
||||
for k in keys
|
||||
for element in util.to_list(self.query[k])
|
||||
)
|
||||
return s
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return self.render_as_string()
|
||||
|
||||
def __copy__(self) -> URL:
|
||||
return self.__class__.create(
|
||||
self.drivername,
|
||||
self.username,
|
||||
self.password,
|
||||
self.host,
|
||||
self.port,
|
||||
self.database,
|
||||
# note this is an immutabledict of str-> str / tuple of str,
|
||||
# also fully immutable. does not require deepcopy
|
||||
self.query,
|
||||
)
|
||||
|
||||
def __deepcopy__(self, memo: Any) -> URL:
|
||||
return self.__copy__()
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(str(self))
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
return (
|
||||
isinstance(other, URL)
|
||||
and self.drivername == other.drivername
|
||||
and self.username == other.username
|
||||
and self.password == other.password
|
||||
and self.host == other.host
|
||||
and self.database == other.database
|
||||
and self.query == other.query
|
||||
and self.port == other.port
|
||||
)
|
||||
|
||||
def __ne__(self, other: Any) -> bool:
|
||||
return not self == other
|
||||
|
||||
def get_backend_name(self) -> str:
|
||||
"""Return the backend name.
|
||||
|
||||
This is the name that corresponds to the database backend in
|
||||
use, and is the portion of the :attr:`_engine.URL.drivername`
|
||||
that is to the left of the plus sign.
|
||||
|
||||
"""
|
||||
if "+" not in self.drivername:
|
||||
return self.drivername
|
||||
else:
|
||||
return self.drivername.split("+")[0]
|
||||
|
||||
def get_driver_name(self) -> str:
|
||||
"""Return the backend name.
|
||||
|
||||
This is the name that corresponds to the DBAPI driver in
|
||||
use, and is the portion of the :attr:`_engine.URL.drivername`
|
||||
that is to the right of the plus sign.
|
||||
|
||||
If the :attr:`_engine.URL.drivername` does not include a plus sign,
|
||||
then the default :class:`_engine.Dialect` for this :class:`_engine.URL`
|
||||
is imported in order to get the driver name.
|
||||
|
||||
"""
|
||||
|
||||
if "+" not in self.drivername:
|
||||
return self.get_dialect().driver
|
||||
else:
|
||||
return self.drivername.split("+")[1]
|
||||
|
||||
def _instantiate_plugins(
|
||||
self, kwargs: Mapping[str, Any]
|
||||
) -> Tuple[URL, List[Any], Dict[str, Any]]:
|
||||
plugin_names = util.to_list(self.query.get("plugin", ()))
|
||||
plugin_names += kwargs.get("plugins", [])
|
||||
|
||||
kwargs = dict(kwargs)
|
||||
|
||||
loaded_plugins = [
|
||||
plugins.load(plugin_name)(self, kwargs)
|
||||
for plugin_name in plugin_names
|
||||
]
|
||||
|
||||
u = self.difference_update_query(["plugin", "plugins"])
|
||||
|
||||
for plugin in loaded_plugins:
|
||||
new_u = plugin.update_url(u)
|
||||
if new_u is not None:
|
||||
u = new_u
|
||||
|
||||
kwargs.pop("plugins", None)
|
||||
|
||||
return u, loaded_plugins, kwargs
|
||||
|
||||
def _get_entrypoint(self) -> Type[Dialect]:
|
||||
"""Return the "entry point" dialect class.
|
||||
|
||||
This is normally the dialect itself except in the case when the
|
||||
returned class implements the get_dialect_cls() method.
|
||||
|
||||
"""
|
||||
if "+" not in self.drivername:
|
||||
name = self.drivername
|
||||
else:
|
||||
name = self.drivername.replace("+", ".")
|
||||
cls = registry.load(name)
|
||||
# check for legacy dialects that
|
||||
# would return a module with 'dialect' as the
|
||||
# actual class
|
||||
if (
|
||||
hasattr(cls, "dialect")
|
||||
and isinstance(cls.dialect, type)
|
||||
and issubclass(cls.dialect, Dialect)
|
||||
):
|
||||
return cls.dialect
|
||||
else:
|
||||
return cast("Type[Dialect]", cls)
|
||||
|
||||
def get_dialect(self, _is_async: bool = False) -> Type[Dialect]:
|
||||
"""Return the SQLAlchemy :class:`_engine.Dialect` class corresponding
|
||||
to this URL's driver name.
|
||||
|
||||
"""
|
||||
entrypoint = self._get_entrypoint()
|
||||
if _is_async:
|
||||
dialect_cls = entrypoint.get_async_dialect_cls(self)
|
||||
else:
|
||||
dialect_cls = entrypoint.get_dialect_cls(self)
|
||||
return dialect_cls
|
||||
|
||||
def translate_connect_args(
|
||||
self, names: Optional[List[str]] = None, **kw: Any
|
||||
) -> Dict[str, Any]:
|
||||
r"""Translate url attributes into a dictionary of connection arguments.
|
||||
|
||||
Returns attributes of this url (`host`, `database`, `username`,
|
||||
`password`, `port`) as a plain dictionary. The attribute names are
|
||||
used as the keys by default. Unset or false attributes are omitted
|
||||
from the final dictionary.
|
||||
|
||||
:param \**kw: Optional, alternate key names for url attributes.
|
||||
|
||||
:param names: Deprecated. Same purpose as the keyword-based alternate
|
||||
names, but correlates the name to the original positionally.
|
||||
"""
|
||||
|
||||
if names is not None:
|
||||
util.warn_deprecated(
|
||||
"The `URL.translate_connect_args.name`s parameter is "
|
||||
"deprecated. Please pass the "
|
||||
"alternate names as kw arguments.",
|
||||
"1.4",
|
||||
)
|
||||
|
||||
translated = {}
|
||||
attribute_names = ["host", "database", "username", "password", "port"]
|
||||
for sname in attribute_names:
|
||||
if names:
|
||||
name = names.pop(0)
|
||||
elif sname in kw:
|
||||
name = kw[sname]
|
||||
else:
|
||||
name = sname
|
||||
if name is not None and getattr(self, sname, False):
|
||||
if sname == "password":
|
||||
translated[name] = str(getattr(self, sname))
|
||||
else:
|
||||
translated[name] = getattr(self, sname)
|
||||
|
||||
return translated
|
||||
|
||||
|
||||
def make_url(name_or_url: Union[str, URL]) -> URL:
|
||||
"""Given a string, produce a new URL instance.
|
||||
|
||||
The format of the URL generally follows `RFC-1738
|
||||
<https://www.ietf.org/rfc/rfc1738.txt>`_, with some exceptions, including
|
||||
that underscores, and not dashes or periods, are accepted within the
|
||||
"scheme" portion.
|
||||
|
||||
If a :class:`.URL` object is passed, it is returned as is.
|
||||
|
||||
.. seealso::
|
||||
|
||||
:ref:`database_urls`
|
||||
|
||||
"""
|
||||
|
||||
if isinstance(name_or_url, str):
|
||||
return _parse_url(name_or_url)
|
||||
elif not isinstance(name_or_url, URL) and not hasattr(
|
||||
name_or_url, "_sqla_is_testing_if_this_is_a_mock_object"
|
||||
):
|
||||
raise exc.ArgumentError(
|
||||
f"Expected string or URL object, got {name_or_url!r}"
|
||||
)
|
||||
else:
|
||||
return name_or_url
|
||||
|
||||
|
||||
def _parse_url(name: str) -> URL:
|
||||
pattern = re.compile(
|
||||
r"""
|
||||
(?P<name>[\w\+]+)://
|
||||
(?:
|
||||
(?P<username>[^:/]*)
|
||||
(?::(?P<password>[^@]*))?
|
||||
@)?
|
||||
(?:
|
||||
(?:
|
||||
\[(?P<ipv6host>[^/\?]+)\] |
|
||||
(?P<ipv4host>[^/:\?]+)
|
||||
)?
|
||||
(?::(?P<port>[^/\?]*))?
|
||||
)?
|
||||
(?:/(?P<database>[^\?]*))?
|
||||
(?:\?(?P<query>.*))?
|
||||
""",
|
||||
re.X,
|
||||
)
|
||||
|
||||
m = pattern.match(name)
|
||||
if m is not None:
|
||||
components = m.groupdict()
|
||||
query: Optional[Dict[str, Union[str, List[str]]]]
|
||||
if components["query"] is not None:
|
||||
query = {}
|
||||
|
||||
for key, value in parse_qsl(components["query"]):
|
||||
if key in query:
|
||||
query[key] = util.to_list(query[key])
|
||||
cast("List[str]", query[key]).append(value)
|
||||
else:
|
||||
query[key] = value
|
||||
else:
|
||||
query = None
|
||||
components["query"] = query
|
||||
|
||||
if components["username"] is not None:
|
||||
components["username"] = unquote(components["username"])
|
||||
|
||||
if components["password"] is not None:
|
||||
components["password"] = unquote(components["password"])
|
||||
|
||||
ipv4host = components.pop("ipv4host")
|
||||
ipv6host = components.pop("ipv6host")
|
||||
components["host"] = ipv4host or ipv6host
|
||||
name = components.pop("name")
|
||||
|
||||
if components["port"]:
|
||||
components["port"] = int(components["port"])
|
||||
|
||||
return URL.create(name, **components) # type: ignore
|
||||
|
||||
else:
|
||||
raise exc.ArgumentError(
|
||||
"Could not parse SQLAlchemy URL from given URL string"
|
||||
)
|
||||
167
venv/Lib/site-packages/sqlalchemy/engine/util.py
Normal file
167
venv/Lib/site-packages/sqlalchemy/engine/util.py
Normal file
@@ -0,0 +1,167 @@
|
||||
# engine/util.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import typing
|
||||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import Optional
|
||||
from typing import TypeVar
|
||||
|
||||
from .. import exc
|
||||
from .. import util
|
||||
from ..util._has_cy import HAS_CYEXTENSION
|
||||
from ..util.typing import Protocol
|
||||
from ..util.typing import Self
|
||||
|
||||
if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
|
||||
from ._py_util import _distill_params_20 as _distill_params_20
|
||||
from ._py_util import _distill_raw_params as _distill_raw_params
|
||||
else:
|
||||
from sqlalchemy.cyextension.util import ( # noqa: F401
|
||||
_distill_params_20 as _distill_params_20,
|
||||
)
|
||||
from sqlalchemy.cyextension.util import ( # noqa: F401
|
||||
_distill_raw_params as _distill_raw_params,
|
||||
)
|
||||
|
||||
_C = TypeVar("_C", bound=Callable[[], Any])
|
||||
|
||||
|
||||
def connection_memoize(key: str) -> Callable[[_C], _C]:
|
||||
"""Decorator, memoize a function in a connection.info stash.
|
||||
|
||||
Only applicable to functions which take no arguments other than a
|
||||
connection. The memo will be stored in ``connection.info[key]``.
|
||||
"""
|
||||
|
||||
@util.decorator
|
||||
def decorated(fn, self, connection): # type: ignore
|
||||
connection = connection.connect()
|
||||
try:
|
||||
return connection.info[key]
|
||||
except KeyError:
|
||||
connection.info[key] = val = fn(self, connection)
|
||||
return val
|
||||
|
||||
return decorated
|
||||
|
||||
|
||||
class _TConsSubject(Protocol):
|
||||
_trans_context_manager: Optional[TransactionalContext]
|
||||
|
||||
|
||||
class TransactionalContext:
|
||||
"""Apply Python context manager behavior to transaction objects.
|
||||
|
||||
Performs validation to ensure the subject of the transaction is not
|
||||
used if the transaction were ended prematurely.
|
||||
|
||||
"""
|
||||
|
||||
__slots__ = ("_outer_trans_ctx", "_trans_subject", "__weakref__")
|
||||
|
||||
_trans_subject: Optional[_TConsSubject]
|
||||
|
||||
def _transaction_is_active(self) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
def _transaction_is_closed(self) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
def _rollback_can_be_called(self) -> bool:
|
||||
"""indicates the object is in a state that is known to be acceptable
|
||||
for rollback() to be called.
|
||||
|
||||
This does not necessarily mean rollback() will succeed or not raise
|
||||
an error, just that there is currently no state detected that indicates
|
||||
rollback() would fail or emit warnings.
|
||||
|
||||
It also does not mean that there's a transaction in progress, as
|
||||
it is usually safe to call rollback() even if no transaction is
|
||||
present.
|
||||
|
||||
.. versionadded:: 1.4.28
|
||||
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def _get_subject(self) -> _TConsSubject:
|
||||
raise NotImplementedError()
|
||||
|
||||
def commit(self) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
def rollback(self) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
def close(self) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
def _trans_ctx_check(cls, subject: _TConsSubject) -> None:
|
||||
trans_context = subject._trans_context_manager
|
||||
if trans_context:
|
||||
if not trans_context._transaction_is_active():
|
||||
raise exc.InvalidRequestError(
|
||||
"Can't operate on closed transaction inside context "
|
||||
"manager. Please complete the context manager "
|
||||
"before emitting further commands."
|
||||
)
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
subject = self._get_subject()
|
||||
|
||||
# none for outer transaction, may be non-None for nested
|
||||
# savepoint, legacy nesting cases
|
||||
trans_context = subject._trans_context_manager
|
||||
self._outer_trans_ctx = trans_context
|
||||
|
||||
self._trans_subject = subject
|
||||
subject._trans_context_manager = self
|
||||
return self
|
||||
|
||||
def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
|
||||
subject = getattr(self, "_trans_subject", None)
|
||||
|
||||
# simplistically we could assume that
|
||||
# "subject._trans_context_manager is self". However, any calling
|
||||
# code that is manipulating __exit__ directly would break this
|
||||
# assumption. alembic context manager
|
||||
# is an example of partial use that just calls __exit__ and
|
||||
# not __enter__ at the moment. it's safe to assume this is being done
|
||||
# in the wild also
|
||||
out_of_band_exit = (
|
||||
subject is None or subject._trans_context_manager is not self
|
||||
)
|
||||
|
||||
if type_ is None and self._transaction_is_active():
|
||||
try:
|
||||
self.commit()
|
||||
except:
|
||||
with util.safe_reraise():
|
||||
if self._rollback_can_be_called():
|
||||
self.rollback()
|
||||
finally:
|
||||
if not out_of_band_exit:
|
||||
assert subject is not None
|
||||
subject._trans_context_manager = self._outer_trans_ctx
|
||||
self._trans_subject = self._outer_trans_ctx = None
|
||||
else:
|
||||
try:
|
||||
if not self._transaction_is_active():
|
||||
if not self._transaction_is_closed():
|
||||
self.close()
|
||||
else:
|
||||
if self._rollback_can_be_called():
|
||||
self.rollback()
|
||||
finally:
|
||||
if not out_of_band_exit:
|
||||
assert subject is not None
|
||||
subject._trans_context_manager = self._outer_trans_ctx
|
||||
self._trans_subject = self._outer_trans_ctx = None
|
||||
Reference in New Issue
Block a user