Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/types/json.py: 58%

1import json 

2 

3import sqlalchemy as sa 

4from sqlalchemy.dialects.postgresql.base import ischema_names 

5 

6try: 

7 from sqlalchemy.dialects.postgresql import JSON 

8 

9 has_postgres_json = True 

10except ImportError: 

11 

12 class PostgresJSONType(sa.types.UserDefinedType): 

13 """ 

14 Text search vector type for postgresql. 

15 """ 

16 

17 def get_col_spec(self): 

18 return 'json' 

19 

20 ischema_names['json'] = PostgresJSONType 

21 has_postgres_json = False 

22 

23 

24class JSONType(sa.types.TypeDecorator): 

25 """ 

26 JSONType offers way of saving JSON data structures to database. On 

27 PostgreSQL the underlying implementation of this data type is 'json' while 

28 on other databases its simply 'text'. 

29 

30 :: 

31 

32 

33 from sqlalchemy_utils import JSONType 

34 

35 

36 class Product(Base): 

37 __tablename__ = 'product' 

38 id = sa.Column(sa.Integer, autoincrement=True) 

39 name = sa.Column(sa.Unicode(50)) 

40 details = sa.Column(JSONType) 

41 

42 

43 product = Product() 

44 product.details = { 

45 'color': 'red', 

46 'type': 'car', 

47 'max-speed': '400 mph' 

48 } 

49 session.commit() 

50 """ 

51 

52 impl = sa.UnicodeText 

53 hashable = False 

54 cache_ok = True 

55 

56 def __init__(self, *args, **kwargs): 

57 super().__init__(*args, **kwargs) 

58 

59 def load_dialect_impl(self, dialect): 

60 if dialect.name == 'postgresql': 

61 # Use the native JSON type. 

62 if has_postgres_json: 

63 return dialect.type_descriptor(JSON()) 

64 else: 

65 return dialect.type_descriptor(PostgresJSONType()) 

66 else: 

67 return dialect.type_descriptor(self.impl) 

68 

69 def process_bind_param(self, value, dialect): 

70 if dialect.name == 'postgresql' and has_postgres_json: 

71 return value 

72 if value is not None: 

73 value = json.dumps(value) 

74 return value 

75 

76 def process_result_value(self, value, dialect): 

77 if dialect.name == 'postgresql': 

78 return value 

79 if value is not None: 

80 value = json.loads(value) 

81 return value