Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/json.py: 38%

1# dialects/mysql/json.py 

2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7from __future__ import annotations 

8 

9from typing import Any 

10from typing import TYPE_CHECKING 

11 

12from ... import types as sqltypes 

13 

14if TYPE_CHECKING: 

15 from ...engine.interfaces import Dialect 

16 from ...sql.type_api import _BindProcessorType 

17 from ...sql.type_api import _LiteralProcessorType 

18 

19 

20class JSON(sqltypes.JSON): 

21 """MySQL JSON type. 

22 

23 MySQL supports JSON as of version 5.7. 

24 MariaDB supports JSON (as an alias for LONGTEXT) as of version 10.2. 

25 

26 :class:`_mysql.JSON` is used automatically whenever the base 

27 :class:`_types.JSON` datatype is used against a MySQL or MariaDB backend. 

28 

29 .. seealso:: 

30 

31 :class:`_types.JSON` - main documentation for the generic 

32 cross-platform JSON datatype. 

33 

34 The :class:`.mysql.JSON` type supports persistence of JSON values 

35 as well as the core index operations provided by :class:`_types.JSON` 

36 datatype, by adapting the operations to render the ``JSON_EXTRACT`` 

37 function at the database level. 

38 

39 """ 

40 

41 pass 

42 

43 

44class _FormatTypeMixin: 

45 def _format_value(self, value: Any) -> str: 

46 raise NotImplementedError() 

47 

48 def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]: 

49 super_proc = self.string_bind_processor(dialect) # type: ignore[attr-defined] # noqa: E501 

50 

51 def process(value: Any) -> Any: 

52 value = self._format_value(value) 

53 if super_proc: 

54 value = super_proc(value) 

55 return value 

56 

57 return process 

58 

59 def literal_processor( 

60 self, dialect: Dialect 

61 ) -> _LiteralProcessorType[Any]: 

62 super_proc = self.string_literal_processor(dialect) # type: ignore[attr-defined] # noqa: E501 

63 

64 def process(value: Any) -> str: 

65 value = self._format_value(value) 

66 if super_proc: 

67 value = super_proc(value) 

68 return value # type: ignore[no-any-return] 

69 

70 return process 

71 

72 

73class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType): 

74 def _format_value(self, value: Any) -> str: 

75 if isinstance(value, int): 

76 formatted_value = "$[%s]" % value 

77 else: 

78 formatted_value = '$."%s"' % value 

79 return formatted_value 

80 

81 

82class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType): 

83 def _format_value(self, value: Any) -> str: 

84 return "$%s" % ( 

85 "".join( 

86 [ 

87 "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem 

88 for elem in value 

89 ] 

90 ) 

91 )