# Copyright (c) 2016, 2022, Oracle and/or its affiliates. All rights reserved.
|
#
|
# This program is free software; you can redistribute it and/or modify
|
# it under the terms of the GNU General Public License, version 2.0, as
|
# published by the Free Software Foundation.
|
#
|
# This program is also distributed with certain software (including
|
# but not limited to OpenSSL) that is licensed under separate terms,
|
# as designated in a particular file or component or in included license
|
# documentation. The authors of MySQL hereby grant you an
|
# additional permission to link the program and your derivative works
|
# with the separately licensed software that they have included with
|
# MySQL.
|
#
|
# Without limiting anything contained in the foregoing, this file,
|
# which is part of MySQL Connector/Python, is also subject to the
|
# Universal FOSS Exception, version 1.0, a copy of which can be found at
|
# http://oss.oracle.com/licenses/universal-foss-exception.
|
#
|
# This program is distributed in the hope that it will be useful, but
|
# WITHOUT ANY WARRANTY; without even the implied warranty of
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
# See the GNU General Public License, version 2.0, for more details.
|
#
|
# You should have received a copy of the GNU General Public License
|
# along with this program; if not, write to the Free Software Foundation, Inc.,
|
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
|
|
"""Implementation of the DbDoc."""
|
|
from __future__ import annotations
|
|
import json
|
|
from typing import Any, Dict, KeysView, Optional, Union
|
|
from .errors import ProgrammingError
|
|
|
class ExprJSONEncoder(json.JSONEncoder):
|
"""A :class:`json.JSONEncoder` subclass, which enables encoding of
|
:class:`mysqlx.ExprParser` objects."""
|
|
def default(self, o: object) -> str:
|
if hasattr(o, "expr"):
|
return f"{o}"
|
# Let the base class default method raise the TypeError
|
return json.JSONEncoder.default(self, o)
|
|
|
class DbDoc:
|
"""Represents a generic document in JSON format.
|
|
Args:
|
value (object): The value can be a JSON string or a dict.
|
|
Raises:
|
ValueError: If ``value`` type is not a basestring or dict.
|
"""
|
|
def __init__(self, value: Union[str, Dict[str, Any]]) -> None:
|
if isinstance(value, dict):
|
self.__dict__ = value
|
elif isinstance(value, str):
|
self.__dict__ = json.loads(value)
|
else:
|
raise ValueError(f"Unable to handle type: {type(value)}")
|
|
def __str__(self) -> str:
|
return self.as_str()
|
|
def __repr__(self) -> str:
|
return repr(self.__dict__)
|
|
def __setitem__(self, index: str, value: Any) -> None:
|
if index == "_id":
|
raise ProgrammingError("Cannot modify _id")
|
self.__dict__[index] = value
|
|
def __getitem__(self, index: str) -> Any:
|
return self.__dict__[index]
|
|
def __contains__(self, item: str) -> bool:
|
return item in self.__dict__
|
|
def copy(self, doc_id: Optional[str] = None) -> DbDoc:
|
"""Returns a new copy of a :class:`mysqlx.DbDoc` object containing the
|
`doc_id` provided. If `doc_id` is not provided, it will be removed from
|
new :class:`mysqlx.DbDoc` object.
|
|
Args:
|
doc_id (Optional[str]): Document ID
|
|
Returns:
|
mysqlx.DbDoc: A new instance of DbDoc containing the _id provided
|
"""
|
new_dict = self.__dict__.copy()
|
if doc_id:
|
new_dict["_id"] = doc_id
|
elif "_id" in new_dict:
|
del new_dict["_id"]
|
return DbDoc(new_dict)
|
|
def keys(self) -> KeysView[str]:
|
"""Returns the keys.
|
|
Returns:
|
`list`: The keys.
|
"""
|
return self.__dict__.keys()
|
|
def as_str(self) -> str:
|
"""Serialize :class:`mysqlx.DbDoc` to a JSON formatted ``str``.
|
|
Returns:
|
str: A JSON formatted ``str`` representation of the document.
|
|
.. versionadded:: 8.0.16
|
"""
|
return json.dumps(self.__dict__, cls=ExprJSONEncoder)
|
