Rebase SqlAlchemy_2.0 Support

Author: nisar123456

.travis.yml

@@ -1,17 +1,18 @@
 language: python
+dist: jammy
 
 matrix:
   include:
-    - python: 2.7
+    - python: 3.8
       env:
-        - TOX_ENV=py27
-    - python: 2.7
+        - TOX_ENV=py38
+    - python: 3.10
       env:
-        - TOX_ENV=pep8
-    - python: 3.6
+        - TOX_ENV=py310
+    - python: 3.12
       env:
-        - TOX_ENV=py36
-    - python: 3.6
+        - TOX_ENV=py312
+    - python: 3.10
       env:
         - TOX_ENV=pep8
 install:

aim/agent/aid/service.py

@@ -235,51 +235,50 @@ def _send_heartbeat(self, aim_ctx):
     def get_vnodes_value(self, aim_ctx):
         dbsession = aim_ctx.store.db_session
         dbsession.get_bind()
-        with dbsession.begin(subtransactions=True):
+        with dbsession.begin():
             aim_consistent_hashring_params_table = sa.Table(
                 'aim_consistent_hashring_params', sa.MetaData(),
                 sa.Column('value', sa.Integer, nullable=False),
                 sa.Column('name', sa.String(16), nullable=False,
                           primary_key=True))
-            query = (sa.select([aim_consistent_hashring_params_table.c.value]).
-                     where(aim_consistent_hashring_params_table.c.name
-                           == 'vnodes'))
+            query = (sa.select(aim_consistent_hashring_params_table.c.value).
+                     where(aim_consistent_hashring_params_table.c.name ==
+                           'vnodes'))
             result = dbsession.execute(query).fetchone()
             vnodes_value = result[0] if result else DEFAULT_VNODES_HASHRING
             return vnodes_value
 
     def _calculate_tenants(self, aim_ctx):
-        with aim_ctx.store.begin(subtransactions=True):
-            # Refresh this agent
-            self.agent = self.manager.get(aim_ctx, self.agent)
-            if not self.single_aid:
-                down_time = self.agent.down_time(aim_ctx)
-                if max(0, down_time or 0) > self.max_down_time:
-                    utils.perform_harakiri(LOG, "Agent has been down for %s "
-                                                "seconds." % down_time)
-                # Get peers
-                agents = [
-                    x for x in self.manager.find(aim_ctx, resource.Agent,
-                                                 admin_state_up=True)
-                    if not x.is_down(aim_ctx)]
-                # Validate agent version
-                if not agents:
-                    return []
-                max_version = max(agents, key=lambda x: x.version).version
-                if self._major_vercompare(self.agent.version, max_version) < 0:
-                    LOG.error("Agent version is outdated: Current %s Required "
-                              "%s" % (self.agent.version, max_version))
-                    return []
-                # Purge outdated agents
-                agents = [x for x in agents if
-                          self._major_vercompare(x.version, max_version) == 0]
-            else:
-                agents = [self.agent]
-            result = self._tenant_assignation_algorithm(aim_ctx, agents)
-            # Store result in DB
-            self.agent.hash_trees = result
-            self.manager.create(aim_ctx, self.agent, overwrite=True)
-            return result
+        # Refresh this agent
+        self.agent = self.manager.get(aim_ctx, self.agent)
+        if not self.single_aid:
+            down_time = self.agent.down_time(aim_ctx)
+            if max(0, down_time or 0) > self.max_down_time:
+                utils.perform_harakiri(LOG, "Agent has been down for %s "
+                                            "seconds." % down_time)
+            # Get peers
+            agents = [
+                x for x in self.manager.find(aim_ctx, resource.Agent,
+                                             admin_state_up=True)
+                if not x.is_down(aim_ctx)]
+            # Validate agent version
+            if not agents:
+                return []
+            max_version = max(agents, key=lambda x: x.version).version
+            if self._major_vercompare(self.agent.version, max_version) < 0:
+                LOG.error("Agent version is outdated: Current %s Required "
+                          "%s" % (self.agent.version, max_version))
+                return []
+            # Purge outdated agents
+            agents = [x for x in agents if
+                      self._major_vercompare(x.version, max_version) == 0]
+        else:
+            agents = [self.agent]
+        result = self._tenant_assignation_algorithm(aim_ctx, agents)
+        # Store result in DB
+        self.agent.hash_trees = result
+        self.manager.create(aim_ctx, self.agent, overwrite=True)
+        return result
 
     def _tenant_assignation_algorithm(self, aim_ctx, agents):
         # TODO(ivar): just randomly hash each tenant to agents for now. This

aim/agent/aid/universes/aci/converter.py

@@ -414,8 +414,8 @@ def l3ext_next_hop_converter(object_dict, otype, helper,
             return []
         for index, attr in enumerate(destination_identity_attributes):
             res_dict[attr] = id[index]
-        if (object_dict.get('type') == 'prefix' and object_dict.get('nhAddr')
-                and object_dict.get('pref')):
+        if (object_dict.get('type') == 'prefix' and object_dict.
+                get('nhAddr') and object_dict.get('pref')):
             res_dict['next_hop_list'] = [{'addr': object_dict['nhAddr'],
                                           'preference': object_dict['pref']}]
         result.append(default_to_resource(res_dict, helper, to_aim=True))

aim/agent/aid/universes/aim_universe.py

@@ -94,8 +94,9 @@ def observe(self, context):
         if utils.get_time() > self._scheduled_recovery:
             for root in served_tenants:
                 self.manager.recover_root_errors(context, root)
-            htdbl.cleanup_zombie_status_objects(context, served_tenants)
-            self.schedule_next_recovery()
+            with context.store.db_session.begin():
+                htdbl.cleanup_zombie_status_objects(context, served_tenants)
+                self.schedule_next_recovery()
         htdbl.catch_up_with_action_log(context.store, served_tenants)
         # REVISIT(ivar): what if a root is marked as needs_reset? we could
         # avoid syncing it altogether
@@ -115,11 +116,9 @@ def get_optimized_state(self, context, other_state,
     def cleanup_state(self, context, key):
         # Only delete if state is still empty. Never remove a tenant if there
         # are leftovers.
-        with context.store.begin(subtransactions=True):
-            # There could still be logs, but they will re-create the
-            # tenants in the next iteration.
-            self.tree_manager.delete_by_root_rn(context, key, if_empty=True)
-        super(AimDbUniverse, self).cleanup_state(context, key)
+        # There could still be logs, but they will re-create the
+        # tenants in the next iteration.
+        self.tree_manager.delete_by_root_rn(context, key, if_empty=True)
 
     def _get_state(self, context, tree=tree_manager.CONFIG_TREE):
         return self.tree_manager.find_changed(
@@ -228,33 +227,32 @@ def _push_resource(self, context, resource, method, monitored):
                         [resource])
                     resource = self._converter.convert(resource)[0]
                     resource.monitored = monitored
-                with context.store.begin(subtransactions=True):
-                    if isinstance(resource, aim_resource.AciRoot):
-                        # Roots should not be created by the
-                        # AIM monitored universe.
-                        # NOTE(ivar): there are contention cases
-                        # where a user might delete a Root object
-                        # right before the AIM Monitored Universe
-                        # pushes an update on it. If we run a
-                        # simple "create overwrite" this would
-                        # re-create the object and AID would keep
-                        # monitoring said root. by only updating
-                        # roots and never creating them, we give
-                        # full control over which trees to monitor
-                        # to the user.
-                        ext = context.store.extract_attributes
-                        obj = self.manager.update(
-                            context, resource, fix_ownership=monitored,
-                            **ext(resource, "other"))
-                        if obj:
-                            # Declare victory for the update
-                            self.creation_succeeded(resource)
-                    else:
-                        self.manager.create(
-                            context, resource, overwrite=True,
-                            fix_ownership=monitored)
-                        # Declare victory for the created object
+                if isinstance(resource, aim_resource.AciRoot):
+                    # Roots should not be created by the
+                    # AIM monitored universe.
+                    # NOTE(ivar): there are contention cases
+                    # where a user might delete a Root object
+                    # right before the AIM Monitored Universe
+                    # pushes an update on it. If we run a
+                    # simple "create overwrite" this would
+                    # re-create the object and AID would keep
+                    # monitoring said root. by only updating
+                    # roots and never creating them, we give
+                    # full control over which trees to monitor
+                    # to the user.
+                    ext = context.store.extract_attributes
+                    obj = self.manager.update(
+                        context, resource, fix_ownership=monitored,
+                        **ext(resource, "other"))
+                    if obj:
+                        # Declare victory for the update
                         self.creation_succeeded(resource)
+                else:
+                    self.manager.create(
+                        context, resource, overwrite=True,
+                        fix_ownership=monitored)
+                    # Declare victory for the created object
+                    self.creation_succeeded(resource)
             else:
                 if isinstance(resource, aim_resource.AciRoot) and monitored:
                     # Monitored Universe doesn't delete Tenant
@@ -264,10 +262,9 @@ def _push_resource(self, context, resource, method, monitored):
                     return
                 if monitored:
                     # Only delete a resource if monitored
-                    with context.store.begin(subtransactions=True):
-                        existing = self.manager.get(context, resource)
-                        if existing and existing.monitored:
-                            self.manager.delete(context, resource)
+                    existing = self.manager.get(context, resource)
+                    if existing and existing.monitored:
+                        self.manager.delete(context, resource)
                 else:
                     self.manager.delete(context, resource)
 

aim/aim_lib/db/model.py

@@ -52,42 +52,59 @@ class CloneL3Out(model_base.Base):
 
 class CloneL3OutManager(object):
 
-    def set(self, context, source, clone):
+    def _set(self, context, source, clone):
         """Store L3outside clone key and the relationship to its source
 
         :param context: AIM context
         :param source: L3Outside AIM resource
         :param clone: L3Outside AIM resource
         :return:
         """
-        with context.db_session.begin(subtransactions=True):
-            obj = CloneL3Out(source_tenant_name=source.tenant_name,
-                             source_name=source.name,
-                             tenant_name=clone.tenant_name,
-                             name=clone.name)
-            context.db_session.add(obj)
+        obj = CloneL3Out(source_tenant_name=source.tenant_name,
+                         source_name=source.name,
+                         tenant_name=clone.tenant_name,
+                         name=clone.name)
+        context.db_session.add(obj)
 
-    def get(self, context, clone):
+    def set(self, context, source, clone):
+        if not context.store.db_session.in_transaction():
+            with context.store.db_session.begin():
+                self._set(context, source, clone)
+        else:
+            self._set(context, source, clone)
+
+    def _get(self, context, clone):
         rows = self._find_query(context, tenant_name=clone.tenant_name,
                                 name=clone.name).all()
         return rows
 
-    def get_clones(self, context, source):
+    def get(self, context, clone):
+        if not context.store.db_session.in_transaction():
+            with context.store.db_session.begin():
+                return self._get(context, clone)
+        return self._get(context, clone)
+
+    def _get_clones(self, context, source):
         """Given a source, find its clones' identity attributes
 
         :param context: AIM context
         :param source: L3Outside AIM resource
         :return: list of tuples where the first position is the clone L3Out
                  tenant_name and the second is its name.
         """
-        with context.db_session.begin(subtransactions=True):
-            result = []
-            db_objs = self._find_query(
-                context, source_tenant_name=source.tenant_name,
-                source_name=source.name).all()
-            for db_obj in db_objs:
-                result.append((db_obj.tenant_name, db_obj.name))
-            return result
+        result = []
+        db_objs = self._find_query(
+            context, source_tenant_name=source.tenant_name,
+            source_name=source.name).all()
+        for db_obj in db_objs:
+            result.append((db_obj.tenant_name, db_obj.name))
+        return result
+
+    def get_clones(self, context, source):
+        if not context.store.db_session.in_transaction():
+            with context.store.db_session.begin():
+                return self._get_clones(context, source)
+        return self._get_clones(context, source)
 
     def _find_query(self, context, **kwargs):
         query = context.db_session.query(CloneL3Out)