Use custom JSONType columns
the JSONType from sqlalchemy_utils is quite brittle as it only does
primitive json.dumps on values. This leads to various sorts of
StorageFailure exceptions in taskflow when, for example, an unserializable
exception bubbles up to the 'failure' field of AtomDetails.
This patch sublclasses the JSONType from sqlalchemy_utils and overrides
two of its methods that do (de)serialization to work via
oslo.serialization functions. They deal with such occurencies much
better, for example, by providing 'str' as a fallback default.
Change-Id: I3b9e9498b155199a4e707006a0cf22cda0567c06
Related-Bug: #1935957
Pavlo Shchelokovskyy authored 2 years ago
Pavlo Shchelokovskyy committed 2 years ago
| 15 | 15 | |
| 16 | 16 | import collections |
| 17 | 17 | |
| 18 | from oslo_serialization import jsonutils | |
| 18 | 19 | from oslo_utils import timeutils |
| 19 | 20 | from oslo_utils import uuidutils |
| 20 | 21 | from sqlalchemy import Table, Column, String, ForeignKey, DateTime, Enum |
| 21 | import sqlalchemy_utils as su | |
| 22 | from sqlalchemy_utils.types import json as json_type | |
| 22 | 23 | |
| 23 | 24 | from taskflow.persistence import models |
| 24 | 25 | from taskflow import states |
| 33 | 34 | VERSION_LENGTH = 64 |
| 34 | 35 | |
| 35 | 36 | |
| 37 | class JSONType(json_type.JSONType): | |
| 38 | """Customized JSONType using oslo.serialization for json operations""" | |
| 39 | ||
| 40 | def process_bind_param(self, value, dialect): | |
| 41 | if dialect.name == 'postgresql' and json_type.has_postgres_json: | |
| 42 | return value | |
| 43 | if value is not None: | |
| 44 | value = jsonutils.dumps(value) | |
| 45 | return value | |
| 46 | ||
| 47 | def process_result_value(self, value, dialect): | |
| 48 | if dialect.name == 'postgresql': | |
| 49 | return value | |
| 50 | if value is not None: | |
| 51 | value = jsonutils.loads(value) | |
| 52 | return value | |
| 53 | ||
| 54 | ||
| 36 | 55 | def fetch(metadata): |
| 37 | 56 | """Returns the master set of table objects (which is also there schema).""" |
| 38 | 57 | logbooks = Table('logbooks', metadata, |
| 40 | 59 | default=timeutils.utcnow), |
| 41 | 60 | Column('updated_at', DateTime, |
| 42 | 61 | onupdate=timeutils.utcnow), |
| 43 | Column('meta', su.JSONType), | |
| 62 | Column('meta', JSONType), | |
| 44 | 63 | Column('name', String(length=NAME_LENGTH)), |
| 45 | 64 | Column('uuid', String(length=UUID_LENGTH), |
| 46 | 65 | primary_key=True, nullable=False, unique=True, |
| 53 | 72 | Column('parent_uuid', String(length=UUID_LENGTH), |
| 54 | 73 | ForeignKey('logbooks.uuid', |
| 55 | 74 | ondelete='CASCADE')), |
| 56 | Column('meta', su.JSONType), | |
| 75 | Column('meta', JSONType), | |
| 57 | 76 | Column('name', String(length=NAME_LENGTH)), |
| 58 | 77 | Column('state', String(length=STATE_LENGTH)), |
| 59 | 78 | Column('uuid', String(length=UUID_LENGTH), |
| 64 | 83 | default=timeutils.utcnow), |
| 65 | 84 | Column('updated_at', DateTime, |
| 66 | 85 | onupdate=timeutils.utcnow), |
| 67 | Column('meta', su.JSONType), | |
| 86 | Column('meta', JSONType), | |
| 68 | 87 | Column('parent_uuid', String(length=UUID_LENGTH), |
| 69 | 88 | ForeignKey('flowdetails.uuid', |
| 70 | 89 | ondelete='CASCADE')), |
| 74 | 93 | Column('uuid', String(length=UUID_LENGTH), |
| 75 | 94 | primary_key=True, nullable=False, unique=True, |
| 76 | 95 | default=uuidutils.generate_uuid), |
| 77 | Column('failure', su.JSONType), | |
| 78 | Column('results', su.JSONType), | |
| 79 | Column('revert_results', su.JSONType), | |
| 80 | Column('revert_failure', su.JSONType), | |
| 96 | Column('failure', JSONType), | |
| 97 | Column('results', JSONType), | |
| 98 | Column('revert_results', JSONType), | |
| 99 | Column('revert_failure', JSONType), | |
| 81 | 100 | Column('atom_type', Enum(*models.ATOM_TYPES, |
| 82 | 101 | name='atom_types')), |
| 83 | 102 | Column('intention', Enum(*states.INTENTIONS, |