[azure][feat]: Add safety read/write access to the graph (#2082)

Author: 1101-1

plugins/azure/fix_plugin_azure/resource/base.py

@@ -3,7 +3,6 @@
 import logging
 from concurrent.futures import Future
 from datetime import datetime, timedelta
-from threading import Lock
 from typing import Any, ClassVar, Dict, Optional, TypeVar, List, Type, Callable, cast
 
 from attr import define, field
@@ -18,6 +17,7 @@
 from fixlib.graph import Graph, EdgeKey
 from fixlib.json_bender import Bender, bend, S, ForallBend, Bend
 from fixlib.threading import ExecutorQueue
+from fixlib.lock import RWLock
 from fixlib.types import Json
 from fixlib.config import current_config
 
@@ -507,8 +507,8 @@ def __init__(
         config: AzureConfig,
         location_lookup: Optional[Dict[str, AzureLocation]] = None,
         location: Optional[AzureLocation] = None,
-        graph_nodes_access: Optional[Lock] = None,
-        graph_edges_access: Optional[Lock] = None,
+        graph_nodes_access: Optional[RWLock] = None,
+        graph_edges_access: Optional[RWLock] = None,
         last_run_started_at: Optional[datetime] = None,
     ) -> None:
         self.graph = graph
@@ -519,8 +519,8 @@ def __init__(
         self.core_feedback = core_feedback
         self.location_lookup = location_lookup or {}
         self.location = location
-        self.graph_nodes_access = graph_nodes_access or Lock()
-        self.graph_edges_access = graph_edges_access or Lock()
+        self.graph_nodes_access = graph_nodes_access or RWLock()
+        self.graph_edges_access = graph_edges_access or RWLock()
         self.name = f"Azure:{subscription.name}"
         self.config = config
         self.last_run_started_at = last_run_started_at
@@ -570,7 +570,7 @@ def node(
         """
         if isinstance(nd := node.get("node"), AzureResource):
             return nd  # type: ignore
-        with self.graph_nodes_access:
+        with self.graph_nodes_access.read_access:
             for n in self.graph:
                 if clazz and not isinstance(n, clazz):
                     continue
@@ -591,7 +591,7 @@ def nodes(
         result: List[AzureResourceType] = []
         if isinstance(nd := node.get("node"), AzureResource):
             result.append(nd)  # type: ignore
-        with self.graph_nodes_access:
+        with self.graph_nodes_access.read_access:
             for n in self.graph:
                 if clazz and not isinstance(n, clazz):
                     continue
@@ -629,7 +629,7 @@ def add_node(self, node: AzureResourceType, source: Optional[Json] = None) -> Op
             node._metadata["provider_link"] = f"https://portal.azure.com/#@/resource{node.id}/overview"
 
         if last_edge_key is not None:
-            with self.graph_nodes_access:
+            with self.graph_nodes_access.write_access:
                 self.graph.add_node(node, source=source or {})
             return node
         else:
@@ -651,7 +651,7 @@ def add_edge(
         if isinstance(from_node, AzureResource) and isinstance(to_n, AzureResource):
             start, end = (to_n, from_node) if reverse else (from_node, to_n)
             log.debug(f"{self.name}: add edge: {start} -> {end} [{edge_type}]")
-            with self.graph_edges_access:
+            with self.graph_edges_access.write_access:
                 return self.graph.add_edge(start, end, edge_type=edge_type)
         return None
 
@@ -677,21 +677,21 @@ def dependant_node(
         if isinstance(from_node, AzureResource) and isinstance(to_n, AzureResource):
             start, end = (to_n, from_node) if reverse else (from_node, to_n)
             log.debug(f"{self.name}: add edge: {start} -> {end} [default]")
-            with self.graph_edges_access:
+            with self.graph_edges_access.write_access:
                 self.graph.add_edge(start, end, edge_type=EdgeType.default)
                 if delete_same_as_default:
                     start, end = end, start
                 log.debug(f"{self.name}: add edge: {end} -> {start} [delete]")
                 self.graph.add_edge(end, start, edge_type=EdgeType.delete)
 
     def resources_of(self, resource_type: Type[AzureResourceType]) -> List[AzureResourceType]:
-        with self.graph_nodes_access:
+        with self.graph_nodes_access.read_access:
             return [n for n in self.graph.nodes if isinstance(n, resource_type)]
 
     def edges_of(
         self, from_type: Type[AzureResource], to_type: Type[AzureResource], edge_type: EdgeType = EdgeType.default
     ) -> List[EdgeKey]:
-        with self.graph_edges_access:
+        with self.graph_edges_access.read_access:
             return [
                 key
                 for (from_node, to_node, key) in self.graph.edges