Apply black formatting to refactored Azure discovery

Author: stefanriegel

azure_discovery/azure_discovery.py

@@ -133,7 +133,164 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
 
         resources = []
 
-        # VMs
+        # Discover resources by type
+        resources.extend(self._discover_vms(rg_name))
+        resources.extend(self._discover_vnets(rg_name))
+        resources.extend(self._discover_load_balancers(rg_name))
+        resources.extend(self._discover_vpn_gateways(rg_name))
+        resources.extend(self._discover_application_gateways(rg_name))
+        resources.extend(self._discover_azure_firewalls(rg_name))
+        resources.extend(self._discover_private_endpoints(rg_name))
+        resources.extend(self._discover_nat_gateways(rg_name))
+        resources.extend(self._discover_route_tables(rg_name))
+        resources.extend(self._discover_public_ip_addresses(rg_name))
+        resources.extend(self._discover_network_security_groups(rg_name))
+        resources.extend(self._discover_express_route_circuits(rg_name))
+        resources.extend(self._discover_dedicated_hosts(rg_name))
+
+        # VNets
+        try:
+            for vnet in self.network_client.virtual_networks.list(rg_name):
+                region = getattr(vnet, "location", "unknown")
+                vnet_name = getattr(vnet, "name", None)
+                if not vnet_name:
+                    self.logger.warning(f"VNet with no name in {rg_name}, skipping subnets.")
+                    continue
+
+                vnet_dict = vars(vnet)
+                formatted_vnet = format_azure_resource(vnet_dict, "vnet", region)
+                resources.append(formatted_vnet)
+
+                # Subnets for this VNet
+                try:
+                    for subnet in self.network_client.subnets.list(rg_name, vnet_name):
+                        subnet_dict = vars(subnet)
+                        formatted_subnet = format_azure_resource(subnet_dict, "subnet", region)
+                        resources.append(formatted_subnet)
+                except Exception as e:
+                    self.logger.warning(f"Error discovering subnets in VNet {vnet_name} in {rg_name}: {e}")
+        except Exception as e:
+            self.logger.warning(f"Error discovering VNets in {rg_name}: {e}")
+
+        # Load Balancers
+        try:
+            for lb in self.network_client.load_balancers.list(rg_name):
+                region = getattr(lb, "location", "unknown")
+                lb_dict = vars(lb)
+                formatted_lb = format_azure_resource(lb_dict, "load_balancer", region)
+                resources.append(formatted_lb)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Load Balancers in {rg_name}: {e}")
+
+        # VPN Gateways
+        try:
+            for vpngw in self.network_client.virtual_network_gateways.list(rg_name):
+                region = getattr(vpngw, "location", "unknown")
+                vpngw_dict = vars(vpngw)
+                formatted_vpngw = format_azure_resource(vpngw_dict, "gateway", region)
+                resources.append(formatted_vpngw)
+        except Exception as e:
+            self.logger.warning(f"Error discovering VPN Gateways in {rg_name}: {e}")
+
+        # Application Gateways
+        try:
+            for appgw in self.network_client.application_gateways.list(rg_name):
+                region = getattr(appgw, "location", "unknown")
+                appgw_dict = vars(appgw)
+                formatted_appgw = format_azure_resource(appgw_dict, "gateway", region)
+                resources.append(formatted_appgw)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Application Gateways in {rg_name}: {e}")
+
+        # Azure Firewalls
+        try:
+            for fw in self.network_client.azure_firewalls.list(rg_name):
+                region = getattr(fw, "location", "unknown")
+                fw_dict = vars(fw)
+                formatted_fw = format_azure_resource(fw_dict, "firewall", region)
+                resources.append(formatted_fw)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Azure Firewalls in {rg_name}: {e}")
+
+        # Private Endpoints
+        try:
+            for pe in self.network_client.private_endpoints.list(rg_name):
+                region = getattr(pe, "location", "unknown")
+                pe_dict = vars(pe)
+                formatted_pe = format_azure_resource(pe_dict, "endpoint", region)
+                resources.append(formatted_pe)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Private Endpoints in {rg_name}: {e}")
+
+        # NAT Gateways
+        try:
+            for natgw in self.network_client.nat_gateways.list(rg_name):
+                region = getattr(natgw, "location", "unknown")
+                natgw_dict = vars(natgw)
+                formatted_natgw = format_azure_resource(natgw_dict, "gateway", region)
+                resources.append(formatted_natgw)
+        except Exception as e:
+            self.logger.warning(f"Error discovering NAT Gateways in {rg_name}: {e}")
+
+        # Route Tables
+        try:
+            for rt in self.network_client.route_tables.list(rg_name):
+                region = getattr(rt, "location", "unknown")
+                rt_dict = vars(rt)
+                formatted_rt = format_azure_resource(rt_dict, "router", region)
+                resources.append(formatted_rt)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Route Tables in {rg_name}: {e}")
+
+        # Public IP Addresses
+        try:
+            for pip in self.network_client.public_ip_addresses.list(rg_name):
+                region = getattr(pip, "location", "unknown")
+                pip_dict = vars(pip)
+                formatted_pip = format_azure_resource(pip_dict, "endpoint", region)
+                resources.append(formatted_pip)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Public IP Addresses in {rg_name}: {e}")
+
+        # Network Security Groups
+        try:
+            for nsg in self.network_client.network_security_groups.list(rg_name):
+                region = getattr(nsg, "location", "unknown")
+                nsg_dict = vars(nsg)
+                formatted_nsg = format_azure_resource(nsg_dict, "switch", region)
+                resources.append(formatted_nsg)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Network Security Groups in {rg_name}: {e}")
+
+        # ExpressRoute Circuits
+        try:
+            for erc in self.network_client.express_route_circuits.list(rg_name):
+                region = getattr(erc, "location", "unknown")
+                erc_dict = vars(erc)
+                formatted_erc = format_azure_resource(erc_dict, "switch", region)
+                resources.append(formatted_erc)
+        except Exception as e:
+            self.logger.warning(f"Error discovering ExpressRoute Circuits in {rg_name}: {e}")
+
+        # Dedicated Hosts
+        try:
+            for host_group in self.compute_client.dedicated_host_groups.list_by_resource_group(rg_name):
+                region = getattr(host_group, "location", "unknown")
+                host_group_name = getattr(host_group, "name", None)
+                if not host_group_name:
+                    continue
+                for host in self.compute_client.dedicated_hosts.list_by_host_group(rg_name, host_group_name):
+                    host_dict = vars(host)
+                    formatted_host = format_azure_resource(host_dict, "server", region)
+                    resources.append(formatted_host)
+        except Exception as e:
+            self.logger.warning(f"Error discovering Dedicated Hosts in {rg_name}: {e}")
+
+        return resources
+
+    def _discover_vms(self, rg_name: str) -> List[Dict]:
+        """Discover Virtual Machines in a resource group."""
+        resources = []
         try:
             for vm in self.compute_client.virtual_machines.list(rg_name):
                 vm_name = getattr(vm, "name", None)
@@ -231,8 +388,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
 
         except Exception as e:
             self.logger.warning(f"Error discovering VMs in {rg_name}: {e}")
+        return resources
 
-        # VNets
+    def _discover_vnets(self, rg_name: str) -> List[Dict]:
+        """Discover Virtual Networks in a resource group."""
+        resources = []
         try:
             for vnet in self.network_client.virtual_networks.list(rg_name):
                 region = getattr(vnet, "location", "unknown")
@@ -255,8 +415,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                     self.logger.warning(f"Error discovering subnets in VNet {vnet_name} in {rg_name}: {e}")
         except Exception as e:
             self.logger.warning(f"Error discovering VNets in {rg_name}: {e}")
+        return resources
 
-        # Load Balancers
+    def _discover_load_balancers(self, rg_name: str) -> List[Dict]:
+        """Discover Load Balancers in a resource group."""
+        resources = []
         try:
             for lb in self.network_client.load_balancers.list(rg_name):
                 region = getattr(lb, "location", "unknown")
@@ -265,8 +428,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_lb)
         except Exception as e:
             self.logger.warning(f"Error discovering Load Balancers in {rg_name}: {e}")
+        return resources
 
-        # VPN Gateways
+    def _discover_vpn_gateways(self, rg_name: str) -> List[Dict]:
+        """Discover VPN Gateways in a resource group."""
+        resources = []
         try:
             for vpngw in self.network_client.virtual_network_gateways.list(rg_name):
                 region = getattr(vpngw, "location", "unknown")
@@ -275,8 +441,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_vpngw)
         except Exception as e:
             self.logger.warning(f"Error discovering VPN Gateways in {rg_name}: {e}")
+        return resources
 
-        # Application Gateways
+    def _discover_application_gateways(self, rg_name: str) -> List[Dict]:
+        """Discover Application Gateways in a resource group."""
+        resources = []
         try:
             for appgw in self.network_client.application_gateways.list(rg_name):
                 region = getattr(appgw, "location", "unknown")
@@ -285,8 +454,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_appgw)
         except Exception as e:
             self.logger.warning(f"Error discovering Application Gateways in {rg_name}: {e}")
+        return resources
 
-        # Azure Firewalls
+    def _discover_azure_firewalls(self, rg_name: str) -> List[Dict]:
+        """Discover Azure Firewalls in a resource group."""
+        resources = []
         try:
             for fw in self.network_client.azure_firewalls.list(rg_name):
                 region = getattr(fw, "location", "unknown")
@@ -295,8 +467,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_fw)
         except Exception as e:
             self.logger.warning(f"Error discovering Azure Firewalls in {rg_name}: {e}")
+        return resources
 
-        # Private Endpoints
+    def _discover_private_endpoints(self, rg_name: str) -> List[Dict]:
+        """Discover Private Endpoints in a resource group."""
+        resources = []
         try:
             for pe in self.network_client.private_endpoints.list(rg_name):
                 region = getattr(pe, "location", "unknown")
@@ -305,8 +480,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_pe)
         except Exception as e:
             self.logger.warning(f"Error discovering Private Endpoints in {rg_name}: {e}")
+        return resources
 
-        # NAT Gateways
+    def _discover_nat_gateways(self, rg_name: str) -> List[Dict]:
+        """Discover NAT Gateways in a resource group."""
+        resources = []
         try:
             for natgw in self.network_client.nat_gateways.list(rg_name):
                 region = getattr(natgw, "location", "unknown")
@@ -315,8 +493,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_natgw)
         except Exception as e:
             self.logger.warning(f"Error discovering NAT Gateways in {rg_name}: {e}")
+        return resources
 
-        # Route Tables
+    def _discover_route_tables(self, rg_name: str) -> List[Dict]:
+        """Discover Route Tables in a resource group."""
+        resources = []
         try:
             for rt in self.network_client.route_tables.list(rg_name):
                 region = getattr(rt, "location", "unknown")
@@ -325,8 +506,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_rt)
         except Exception as e:
             self.logger.warning(f"Error discovering Route Tables in {rg_name}: {e}")
+        return resources
 
-        # Public IP Addresses
+    def _discover_public_ip_addresses(self, rg_name: str) -> List[Dict]:
+        """Discover Public IP Addresses in a resource group."""
+        resources = []
         try:
             for pip in self.network_client.public_ip_addresses.list(rg_name):
                 region = getattr(pip, "location", "unknown")
@@ -335,8 +519,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_pip)
         except Exception as e:
             self.logger.warning(f"Error discovering Public IP Addresses in {rg_name}: {e}")
+        return resources
 
-        # Network Security Groups
+    def _discover_network_security_groups(self, rg_name: str) -> List[Dict]:
+        """Discover Network Security Groups in a resource group."""
+        resources = []
         try:
             for nsg in self.network_client.network_security_groups.list(rg_name):
                 region = getattr(nsg, "location", "unknown")
@@ -345,8 +532,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_nsg)
         except Exception as e:
             self.logger.warning(f"Error discovering Network Security Groups in {rg_name}: {e}")
+        return resources
 
-        # ExpressRoute Circuits
+    def _discover_express_route_circuits(self, rg_name: str) -> List[Dict]:
+        """Discover ExpressRoute Circuits in a resource group."""
+        resources = []
         try:
             for erc in self.network_client.express_route_circuits.list(rg_name):
                 region = getattr(erc, "location", "unknown")
@@ -355,8 +545,11 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                 resources.append(formatted_erc)
         except Exception as e:
             self.logger.warning(f"Error discovering ExpressRoute Circuits in {rg_name}: {e}")
+        return resources
 
-        # Dedicated Hosts
+    def _discover_dedicated_hosts(self, rg_name: str) -> List[Dict]:
+        """Discover Dedicated Hosts in a resource group."""
+        resources = []
         try:
             for host_group in self.compute_client.dedicated_host_groups.list_by_resource_group(rg_name):
                 region = getattr(host_group, "location", "unknown")
@@ -369,7 +562,6 @@ def _discover_resource_group_resources(self, resource_group) -> List[Dict]:
                     resources.append(formatted_host)
         except Exception as e:
             self.logger.warning(f"Error discovering Dedicated Hosts in {rg_name}: {e}")
-
         return resources
 
     def _discover_azure_dns_zones_and_records(self) -> List[Dict]: