#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: ignore-errors
+from __future__ import annotations
+
+from typing import Any
+from typing import Callable
+from typing import List
+from typing import Optional
+from typing import TYPE_CHECKING
+from typing import Union
from .array import ARRAY
from .array import array as _pg_array
from .operators import PATH_MATCH
from ... import types as sqltypes
from ...sql import cast
+from ...sql._typing import _T
+
+if TYPE_CHECKING:
+ from ...engine.interfaces import Dialect
+ from ...sql.elements import ColumnElement
+ from ...sql.type_api import _BindProcessorType
+ from ...sql.type_api import _LiteralProcessorType
+ from ...sql.type_api import TypeEngine
__all__ = ("JSON", "JSONB")
class JSONPathType(sqltypes.JSON.JSONPathType):
- def _processor(self, dialect, super_proc):
- def process(value):
+ def _processor(
+ self, dialect: Dialect, super_proc: Optional[Callable[[Any], Any]]
+ ) -> Callable[[Any], Any]:
+ def process(value: Any) -> Any:
if isinstance(value, str):
# If it's already a string assume that it's in json path
# format. This allows using cast with json paths literals
return process
- def bind_processor(self, dialect):
- return self._processor(dialect, self.string_bind_processor(dialect))
+ def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]:
+ return self._processor(dialect, self.string_bind_processor(dialect)) # type: ignore[return-value] # noqa: E501
- def literal_processor(self, dialect):
- return self._processor(dialect, self.string_literal_processor(dialect))
+ def literal_processor(
+ self, dialect: Dialect
+ ) -> _LiteralProcessorType[Any]:
+ return self._processor(dialect, self.string_literal_processor(dialect)) # type: ignore[return-value] # noqa: E501
class JSONPATH(JSONPathType):
""" # noqa
render_bind_cast = True
- astext_type = sqltypes.Text()
+ astext_type: TypeEngine[str] = sqltypes.Text()
- def __init__(self, none_as_null=False, astext_type=None):
+ def __init__(
+ self,
+ none_as_null: bool = False,
+ astext_type: Optional[TypeEngine[str]] = None,
+ ):
"""Construct a :class:`_types.JSON` type.
:param none_as_null: if True, persist the value ``None`` as a
if astext_type is not None:
self.astext_type = astext_type
- class Comparator(sqltypes.JSON.Comparator):
+ class Comparator(sqltypes.JSON.Comparator[_T]):
"""Define comparison operations for :class:`_types.JSON`."""
+ type: JSON
+
@property
- def astext(self):
+ def astext(self) -> ColumnElement[str]:
"""On an indexed expression, use the "astext" (e.g. "->>")
conversion when rendered in SQL.
"""
if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType):
- return self.expr.left.operate(
+ return self.expr.left.operate( # type: ignore[no-any-return]
JSONPATH_ASTEXT,
self.expr.right,
result_type=self.type.astext_type,
)
else:
- return self.expr.left.operate(
+ return self.expr.left.operate( # type: ignore[no-any-return]
ASTEXT, self.expr.right, result_type=self.type.astext_type
)
__visit_name__ = "JSONB"
- class Comparator(JSON.Comparator):
+ class Comparator(JSON.Comparator[_T]):
"""Define comparison operations for :class:`_types.JSON`."""
- def has_key(self, other):
+ type: JSONB
+
+ def has_key(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test for presence of a key (equivalent of
the ``?`` operator). Note that the key may be a SQLA expression.
"""
return self.operate(HAS_KEY, other, result_type=sqltypes.Boolean)
- def has_all(self, other):
+ def has_all(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test for presence of all keys in jsonb
(equivalent of the ``?&`` operator)
"""
return self.operate(HAS_ALL, other, result_type=sqltypes.Boolean)
- def has_any(self, other):
+ def has_any(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test for presence of any key in jsonb
(equivalent of the ``?|`` operator)
"""
return self.operate(HAS_ANY, other, result_type=sqltypes.Boolean)
- def contains(self, other, **kwargs):
+ def contains(self, other: Any, **kwargs: Any) -> ColumnElement[bool]:
"""Boolean expression. Test if keys (or array) are a superset
of/contained the keys of the argument jsonb expression
(equivalent of the ``@>`` operator).
"""
return self.operate(CONTAINS, other, result_type=sqltypes.Boolean)
- def contained_by(self, other):
+ def contained_by(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test if keys are a proper subset of the
keys of the argument jsonb expression
(equivalent of the ``<@`` operator).
CONTAINED_BY, other, result_type=sqltypes.Boolean
)
- def delete_path(self, array):
+ def delete_path(
+ self, array: Union[List[str], _pg_array[str]]
+ ) -> ColumnElement[JSONB]:
"""JSONB expression. Deletes field or array element specified in
the argument array (equivalent of the ``#-`` operator).
.. versionadded:: 2.0
"""
if not isinstance(array, _pg_array):
- array = _pg_array(array)
+ array = _pg_array(array) # type: ignore[no-untyped-call]
right_side = cast(array, ARRAY(sqltypes.TEXT))
return self.operate(DELETE_PATH, right_side, result_type=JSONB)
- def path_exists(self, other):
+ def path_exists(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test for presence of item given by the
argument JSONPath expression (equivalent of the ``@?`` operator).
PATH_EXISTS, other, result_type=sqltypes.Boolean
)
- def path_match(self, other):
+ def path_match(self, other: Any) -> ColumnElement[bool]:
"""Boolean expression. Test if JSONPath predicate given by the
argument JSONPath expression matches
(equivalent of the ``@@`` operator).
from .schema import MetaData
from .type_api import _BindProcessorType
from .type_api import _ComparatorFactory
+ from .type_api import _LiteralProcessorType
from .type_api import _MatchedOnType
from .type_api import _ResultProcessorType
from ..engine.interfaces import Dialect
_integer = Integer()
_string = String()
- def string_bind_processor(self, dialect):
+ def string_bind_processor(
+ self, dialect: Dialect
+ ) -> Optional[_BindProcessorType[str]]:
return self._string._cached_bind_processor(dialect)
- def string_literal_processor(self, dialect):
+ def string_literal_processor(
+ self, dialect: Dialect
+ ) -> Optional[_LiteralProcessorType[str]]:
return self._string._cached_literal_processor(dialect)
- def bind_processor(self, dialect):
+ def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]:
int_processor = self._integer._cached_bind_processor(dialect)
string_processor = self.string_bind_processor(dialect)
- def process(value):
+ def process(value: Optional[Any]) -> Any:
if int_processor and isinstance(value, int):
value = int_processor(value)
elif string_processor and isinstance(value, str):
return process
- def literal_processor(self, dialect):
+ def literal_processor(
+ self, dialect: Dialect
+ ) -> _LiteralProcessorType[Any]:
int_processor = self._integer._cached_literal_processor(dialect)
string_processor = self.string_literal_processor(dialect)
- def process(value):
+ def process(value: Optional[Any]) -> Any:
if int_processor and isinstance(value, int):
value = int_processor(value)
elif string_processor and isinstance(value, str):
__slots__ = ()
+ type: JSON
+
def _setup_getitem(self, index):
if not isinstance(index, str) and isinstance(
index, collections_abc.Sequence
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
