Coverage for watcher/objects/base.py: 100%
61 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-17 12:22 +0000
1# Copyright 2013 IBM Corp.
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may
4# not use this file except in compliance with the License. You may obtain
5# a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations
13# under the License.
15"""Watcher common internal object model"""
17from oslo_utils import versionutils
18from oslo_versionedobjects import base as ovo_base
19from oslo_versionedobjects import fields as ovo_fields
21from watcher import objects
23remotable_classmethod = ovo_base.remotable_classmethod
24remotable = ovo_base.remotable
27def get_attrname(name):
28 """Return the mangled name of the attribute's underlying storage."""
29 # FIXME(danms): This is just until we use o.vo's class properties
30 # and object base.
31 return '_obj_' + name
34class WatcherObjectRegistry(ovo_base.VersionedObjectRegistry):
35 notification_classes = []
37 def registration_hook(self, cls, index):
38 # NOTE(danms): This is called when an object is registered,
39 # and is responsible for maintaining watcher.objects.$OBJECT
40 # as the highest-versioned implementation of a given object.
41 version = versionutils.convert_version_to_tuple(cls.VERSION)
42 if not hasattr(objects, cls.obj_name()):
43 setattr(objects, cls.obj_name(), cls)
44 else:
45 cur_version = versionutils.convert_version_to_tuple(
46 getattr(objects, cls.obj_name()).VERSION)
47 if version >= cur_version:
48 setattr(objects, cls.obj_name(), cls)
50 @classmethod
51 def register_notification(cls, notification_cls):
52 """Register a class as notification.
54 Use only to register concrete notification or payload classes,
55 do not register base classes intended for inheritance only.
56 """
57 cls.register_if(False)(notification_cls)
58 cls.notification_classes.append(notification_cls)
59 return notification_cls
61 @classmethod
62 def register_notification_objects(cls):
63 """Register previously decorated notification as normal ovos.
65 This is not intended for production use but only for testing and
66 document generation purposes.
67 """
68 for notification_cls in cls.notification_classes:
69 cls.register(notification_cls)
72class WatcherObject(ovo_base.VersionedObject):
73 """Base class and object factory.
75 This forms the base of all objects that can be remoted or instantiated
76 via RPC. Simply defining a class that inherits from this base class
77 will make it remotely instantiatable. Objects should implement the
78 necessary "get" classmethod routines as well as "save" object methods
79 as appropriate.
80 """
82 OBJ_SERIAL_NAMESPACE = 'watcher_object'
83 OBJ_PROJECT_NAMESPACE = 'watcher'
85 def as_dict(self):
86 return {
87 k: getattr(self, k) for k in self.fields
88 if self.obj_attr_is_set(k)}
91class WatcherObjectDictCompat(ovo_base.VersionedObjectDictCompat):
92 pass
95class WatcherComparableObject(ovo_base.ComparableVersionedObject):
96 pass
99class WatcherPersistentObject(object):
100 """Mixin class for Persistent objects.
102 This adds the fields that we use in common for all persistent objects.
103 """
104 fields = {
105 'created_at': ovo_fields.DateTimeField(nullable=True),
106 'updated_at': ovo_fields.DateTimeField(nullable=True),
107 'deleted_at': ovo_fields.DateTimeField(nullable=True),
108 }
110 # Mapping between the object field name and a 2-tuple pair composed of
111 # its object type (e.g. objects.RelatedObject) and the name of the
112 # model field related ID (or UUID) foreign key field.
113 # e.g.:
114 #
115 # fields = {
116 # # [...]
117 # 'related_object_id': fields.IntegerField(), # Foreign key
118 # 'related_object': wfields.ObjectField('RelatedObject'),
119 # }
120 # {'related_object': (objects.RelatedObject, 'related_object_id')}
121 object_fields = {}
123 def obj_refresh(self, loaded_object):
124 """Applies updates for objects that inherit from base.WatcherObject.
126 Checks for updated attributes in an object. Updates are applied from
127 the loaded object column by column in comparison with the current
128 object.
129 """
130 fields = (field for field in self.fields
131 if field not in self.object_fields)
132 for field in fields:
133 if (self.obj_attr_is_set(field) and
134 self[field] != loaded_object[field]):
135 self[field] = loaded_object[field]
137 @staticmethod
138 def _from_db_object(obj, db_object, eager=False):
139 """Converts a database entity to a formal object.
141 :param obj: An object of the class.
142 :param db_object: A DB model of the object
143 :param eager: Enable the loading of object fields (Default: False)
144 :return: The object of the class with the database entity added
146 """
147 obj_class = type(obj)
148 object_fields = obj_class.object_fields
150 for field in obj.fields:
151 if field not in object_fields:
152 obj[field] = db_object[field]
154 if eager:
155 # Load object fields
156 context = obj._context
157 loadable_fields = (
158 (obj_field, related_obj_cls, rel_id)
159 for obj_field, (related_obj_cls, rel_id)
160 in object_fields.items()
161 if obj[rel_id]
162 )
163 for obj_field, related_obj_cls, rel_id in loadable_fields:
164 if getattr(db_object, obj_field, None) and obj[rel_id]:
165 # The object field data was eagerly loaded alongside
166 # the main object data
167 obj[obj_field] = related_obj_cls._from_db_object(
168 related_obj_cls(context), db_object[obj_field])
169 else:
170 # The object field data wasn't loaded yet
171 obj[obj_field] = related_obj_cls.get(context, obj[rel_id])
173 obj.obj_reset_changes()
174 return obj
177class WatcherObjectSerializer(ovo_base.VersionedObjectSerializer):
178 # Base class to use for object hydration
179 OBJ_BASE_CLASS = WatcherObject