Source code for heat.objects.resource_properties_data

#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""ResourcePropertiesData object."""

from oslo_config import cfg
from oslo_serialization import jsonutils
from oslo_versionedobjects import base
from oslo_versionedobjects import fields

from heat.common import crypt
from heat.db.sqlalchemy import api as db_api
from heat.objects import fields as heat_fields


[docs]class ResourcePropertiesData(
    base.VersionedObject,
    base.VersionedObjectDictCompat,
    base.ComparableVersionedObject,
):
    fields = {
        'id': fields.IntegerField(),
        'data': heat_fields.JsonField(nullable=True),
        'created_at': fields.DateTimeField(read_only=True),
        'updated_at': fields.DateTimeField(nullable=True),
    }

    @staticmethod
    def _from_db_object(rpd, context, db_rpd, data_unencrypted=None):
        # The data_unencrypted field allows us to avoid an extra
        # decrypt operation, e.g. when called from create().
        for field in rpd.fields:
            rpd[field] = db_rpd[field]
        if data_unencrypted:  # save a little (decryption) processing
            rpd['data'] = data_unencrypted
        elif db_rpd['encrypted'] and rpd['data'] is not None:
            rpd['data'] = crypt.decrypted_dict(rpd['data'])

        # TODO(cwolfe) setting the context here should go away, that
        # should have been done with the initialisation of the rpd
        # object. For now, maintaining consistency with other
        # _from_db_object methods.
        rpd._context = context
        rpd.obj_reset_changes()
        return rpd

[docs]    @classmethod
    def create_or_update(cls, context, data, rpd_id=None):
        properties_data_encrypted, properties_data = \
            ResourcePropertiesData.encrypt_properties_data(data)
        values = {'encrypted': properties_data_encrypted,
                  'data': properties_data}
        db_obj = db_api.resource_prop_data_create_or_update(
            context, values, rpd_id)
        return cls._from_db_object(cls(), context, db_obj, data)

[docs]    @classmethod
    def create(cls, context, data):
        return ResourcePropertiesData.create_or_update(context, data)

[docs]    @staticmethod
    def encrypt_properties_data(data):
        if cfg.CONF.encrypt_parameters_and_properties and data:
            result = {}
            for prop_name, prop_value in data.items():
                prop_string = jsonutils.dumps(prop_value)
                encrypted_value = crypt.encrypt(prop_string)
                result[prop_name] = encrypted_value
            return (True, result)
        return (False, data)

[docs]    @staticmethod
    def get_by_id(context, id):
        db_ref = db_api.resource_prop_data_get(context, id)
        return ResourcePropertiesData._from_db_object(
            ResourcePropertiesData(context=context), context, db_ref)