aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py')
-rw-r--r--.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py367
1 files changed, 367 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py b/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py
new file mode 100644
index 00000000..663be8b7
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/json.py
@@ -0,0 +1,367 @@
+# dialects/postgresql/json.py
+# Copyright (C) 2005-2025 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
+
+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 ASTEXT
+from .operators import CONTAINED_BY
+from .operators import CONTAINS
+from .operators import DELETE_PATH
+from .operators import HAS_ALL
+from .operators import HAS_ANY
+from .operators import HAS_KEY
+from .operators import JSONPATH_ASTEXT
+from .operators import PATH_EXISTS
+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: 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 value
+ elif value:
+ # If it's already a string assume that it's in json path
+ # format. This allows using cast with json paths literals
+ value = "{%s}" % (", ".join(map(str, value)))
+ else:
+ value = "{}"
+ if super_proc:
+ value = super_proc(value)
+ return value
+
+ return process
+
+ 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: Dialect
+ ) -> _LiteralProcessorType[Any]:
+ return self._processor(dialect, self.string_literal_processor(dialect)) # type: ignore[return-value] # noqa: E501
+
+
+class JSONPATH(JSONPathType):
+ """JSON Path Type.
+
+ This is usually required to cast literal values to json path when using
+ json search like function, such as ``jsonb_path_query_array`` or
+ ``jsonb_path_exists``::
+
+ stmt = sa.select(
+ sa.func.jsonb_path_query_array(
+ table.c.jsonb_col, cast("$.address.id", JSONPATH)
+ )
+ )
+
+ """
+
+ __visit_name__ = "JSONPATH"
+
+
+class JSON(sqltypes.JSON):
+ """Represent the PostgreSQL JSON type.
+
+ :class:`_postgresql.JSON` is used automatically whenever the base
+ :class:`_types.JSON` datatype is used against a PostgreSQL backend,
+ however base :class:`_types.JSON` datatype does not provide Python
+ accessors for PostgreSQL-specific comparison methods such as
+ :meth:`_postgresql.JSON.Comparator.astext`; additionally, to use
+ PostgreSQL ``JSONB``, the :class:`_postgresql.JSONB` datatype should
+ be used explicitly.
+
+ .. seealso::
+
+ :class:`_types.JSON` - main documentation for the generic
+ cross-platform JSON datatype.
+
+ The operators provided by the PostgreSQL version of :class:`_types.JSON`
+ include:
+
+ * Index operations (the ``->`` operator)::
+
+ data_table.c.data["some key"]
+
+ data_table.c.data[5]
+
+ * Index operations returning text
+ (the ``->>`` operator)::
+
+ data_table.c.data["some key"].astext == "some value"
+
+ Note that equivalent functionality is available via the
+ :attr:`.JSON.Comparator.as_string` accessor.
+
+ * Index operations with CAST
+ (equivalent to ``CAST(col ->> ['some key'] AS <type>)``)::
+
+ data_table.c.data["some key"].astext.cast(Integer) == 5
+
+ Note that equivalent functionality is available via the
+ :attr:`.JSON.Comparator.as_integer` and similar accessors.
+
+ * Path index operations (the ``#>`` operator)::
+
+ data_table.c.data[("key_1", "key_2", 5, ..., "key_n")]
+
+ * Path index operations returning text (the ``#>>`` operator)::
+
+ data_table.c.data[
+ ("key_1", "key_2", 5, ..., "key_n")
+ ].astext == "some value"
+
+ Index operations return an expression object whose type defaults to
+ :class:`_types.JSON` by default,
+ so that further JSON-oriented instructions
+ may be called upon the result type.
+
+ Custom serializers and deserializers are specified at the dialect level,
+ that is using :func:`_sa.create_engine`. The reason for this is that when
+ using psycopg2, the DBAPI only allows serializers at the per-cursor
+ or per-connection level. E.g.::
+
+ engine = create_engine(
+ "postgresql+psycopg2://scott:tiger@localhost/test",
+ json_serializer=my_serialize_fn,
+ json_deserializer=my_deserialize_fn,
+ )
+
+ When using the psycopg2 dialect, the json_deserializer is registered
+ against the database using ``psycopg2.extras.register_default_json``.
+
+ .. seealso::
+
+ :class:`_types.JSON` - Core level JSON type
+
+ :class:`_postgresql.JSONB`
+
+ """ # noqa
+
+ render_bind_cast = True
+ astext_type: TypeEngine[str] = sqltypes.Text()
+
+ 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
+ SQL NULL value, not the JSON encoding of ``null``. Note that
+ when this flag is False, the :func:`.null` construct can still
+ be used to persist a NULL value::
+
+ from sqlalchemy import null
+
+ conn.execute(table.insert(), {"data": null()})
+
+ .. seealso::
+
+ :attr:`_types.JSON.NULL`
+
+ :param astext_type: the type to use for the
+ :attr:`.JSON.Comparator.astext`
+ accessor on indexed attributes. Defaults to :class:`_types.Text`.
+
+ """
+ super().__init__(none_as_null=none_as_null)
+ if astext_type is not None:
+ self.astext_type = astext_type
+
+ class Comparator(sqltypes.JSON.Comparator[_T]):
+ """Define comparison operations for :class:`_types.JSON`."""
+
+ type: JSON
+
+ @property
+ def astext(self) -> ColumnElement[str]:
+ """On an indexed expression, use the "astext" (e.g. "->>")
+ conversion when rendered in SQL.
+
+ E.g.::
+
+ select(data_table.c.data["some key"].astext)
+
+ .. seealso::
+
+ :meth:`_expression.ColumnElement.cast`
+
+ """
+ if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType):
+ 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( # type: ignore[no-any-return]
+ ASTEXT, self.expr.right, result_type=self.type.astext_type
+ )
+
+ comparator_factory = Comparator
+
+
+class JSONB(JSON):
+ """Represent the PostgreSQL JSONB type.
+
+ The :class:`_postgresql.JSONB` type stores arbitrary JSONB format data,
+ e.g.::
+
+ data_table = Table(
+ "data_table",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", JSONB),
+ )
+
+ with engine.connect() as conn:
+ conn.execute(
+ data_table.insert(), data={"key1": "value1", "key2": "value2"}
+ )
+
+ The :class:`_postgresql.JSONB` type includes all operations provided by
+ :class:`_types.JSON`, including the same behaviors for indexing
+ operations.
+ It also adds additional operators specific to JSONB, including
+ :meth:`.JSONB.Comparator.has_key`, :meth:`.JSONB.Comparator.has_all`,
+ :meth:`.JSONB.Comparator.has_any`, :meth:`.JSONB.Comparator.contains`,
+ :meth:`.JSONB.Comparator.contained_by`,
+ :meth:`.JSONB.Comparator.delete_path`,
+ :meth:`.JSONB.Comparator.path_exists` and
+ :meth:`.JSONB.Comparator.path_match`.
+
+ Like the :class:`_types.JSON` type, the :class:`_postgresql.JSONB`
+ type does not detect
+ in-place changes when used with the ORM, unless the
+ :mod:`sqlalchemy.ext.mutable` extension is used.
+
+ Custom serializers and deserializers
+ are shared with the :class:`_types.JSON` class,
+ using the ``json_serializer``
+ and ``json_deserializer`` keyword arguments. These must be specified
+ at the dialect level using :func:`_sa.create_engine`. When using
+ psycopg2, the serializers are associated with the jsonb type using
+ ``psycopg2.extras.register_default_jsonb`` on a per-connection basis,
+ in the same way that ``psycopg2.extras.register_default_json`` is used
+ to register these handlers with the json type.
+
+ .. seealso::
+
+ :class:`_types.JSON`
+
+ """
+
+ __visit_name__ = "JSONB"
+
+ class Comparator(JSON.Comparator[_T]):
+ """Define comparison operations for :class:`_types.JSON`."""
+
+ 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: 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: 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: 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).
+
+ kwargs may be ignored by this operator but are required for API
+ conformance.
+ """
+ return self.operate(CONTAINS, other, result_type=sqltypes.Boolean)
+
+ 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).
+ """
+ return self.operate(
+ CONTAINED_BY, other, result_type=sqltypes.Boolean
+ )
+
+ 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).
+
+ The input may be a list of strings that will be coerced to an
+ ``ARRAY`` or an instance of :meth:`_postgres.array`.
+
+ .. versionadded:: 2.0
+ """
+ if not isinstance(array, _pg_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: Any) -> ColumnElement[bool]:
+ """Boolean expression. Test for presence of item given by the
+ argument JSONPath expression (equivalent of the ``@?`` operator).
+
+ .. versionadded:: 2.0
+ """
+ return self.operate(
+ PATH_EXISTS, other, result_type=sqltypes.Boolean
+ )
+
+ 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).
+
+ Only the first item of the result is taken into account.
+
+ .. versionadded:: 2.0
+ """
+ return self.operate(
+ PATH_MATCH, other, result_type=sqltypes.Boolean
+ )
+
+ comparator_factory = Comparator