refactor: Use PEP585 type hints for standard collections

Author: gfieni

src/powerapi/cli/config_validator.py

@@ -30,8 +30,6 @@
 import logging
 import os
 
-from typing import Dict
-
 from powerapi.exception import MissingArgumentException, NotAllowedArgumentValueException, FileDoesNotExistException, \
     UnexistingActorException
 
@@ -42,7 +40,7 @@ class ConfigValidator:
     """
 
     @staticmethod
-    def validate(config: Dict):
+    def validate(config: dict):
         """
         Validate powerapi config and initialize missing default values
         """
@@ -101,7 +99,7 @@ def validate(config: Dict):
         ConfigValidator._validate_input(config)
 
     @staticmethod
-    def _validate_input(config: Dict):
+    def _validate_input(config: dict):
         """
         Check that csv input type has files that exist
         """
@@ -118,7 +116,7 @@ def _validate_input(config: Dict):
                         raise FileDoesNotExistException(file_name=file_name)
 
     @staticmethod
-    def _validate_binding(config: Dict):
+    def _validate_binding(config: dict):
         """
         Check that defined bindings use existing actors defined by the configuration
         """

src/powerapi/cli/generator.py

@@ -30,7 +30,7 @@
 import logging
 import os
 import sys
-from typing import Dict, Type, Callable
+from typing import Callable
 
 from powerapi.actor import Actor
 from powerapi.database import MongoDB, CsvDB, OpenTSDB, SocketDB, PrometheusDB, VirtioFSDB, FileDB
@@ -93,7 +93,7 @@ class Generator:
     def __init__(self, component_group_name):
         self.component_group_name = component_group_name
 
-    def generate(self, main_config: dict) -> Dict[str, Type[Actor]]:
+    def generate(self, main_config: dict) -> dict[str, type[Actor]]:
         """
         Generate an actor class and actor start message from config dict
         """
@@ -115,7 +115,7 @@ def generate(self, main_config: dict) -> Dict[str, Type[Actor]]:
 
         return actors
 
-    def _gen_actor(self, component_config: dict, main_config: dict, component_name: str) -> Type[Actor]:
+    def _gen_actor(self, component_config: dict, main_config: dict, component_name: str) -> type[Actor]:
         raise NotImplementedError()
 
 
@@ -213,7 +213,7 @@ def remove_db_factory(self, database_name: str):
             raise DatabaseNameDoesNotExist(database_name)
         del self.db_factory[database_name]
 
-    def add_report_class(self, model_name: str, report_class: Type[Report]):
+    def add_report_class(self, model_name: str, report_class: type[Report]):
         """
         add a report class to generator
         """
@@ -287,7 +287,7 @@ class ProcessorGenerator(Generator):
     Generator that initialises the processor from config
     """
 
-    def __init__(self, component_group_name: str, processor_factory: Dict[str, Callable[[Dict], ProcessorActor]] = None):
+    def __init__(self, component_group_name: str, processor_factory: dict[str, Callable[[dict], ProcessorActor]] = None):
         Generator.__init__(self, component_group_name)
 
         self.processor_factory = processor_factory
@@ -331,7 +331,7 @@ def __init__(self):
         super().__init__('pre-processor', self._get_default_processor_factories())
 
     @staticmethod
-    def _k8s_pre_processor_factory(processor_config: Dict) -> K8sPreProcessorActor:
+    def _k8s_pre_processor_factory(processor_config: dict) -> K8sPreProcessorActor:
         """
         Kubernetes pre-processor actor factory.
         :param processor_config: Pre-Processor configuration
@@ -345,7 +345,7 @@ def _k8s_pre_processor_factory(processor_config: Dict) -> K8sPreProcessorActor:
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
         return K8sPreProcessorActor(name, [], target_actors_name, api_mode, api_host, api_key, level_logger)
 
-    def _get_default_processor_factories(self) -> Dict[str, Callable[[Dict], ProcessorActor]]:
+    def _get_default_processor_factories(self) -> dict[str, Callable[[dict], ProcessorActor]]:
         """
         Return the default pre-processors factory.
         """
@@ -363,5 +363,5 @@ def __init__(self):
         ProcessorGenerator.__init__(self, 'post-processor', self._get_default_processor_factories())
 
     @staticmethod
-    def _get_default_processor_factories() -> Dict[str, Callable[[Dict], ProcessorActor]]:
+    def _get_default_processor_factories() -> dict[str, Callable[[dict], ProcessorActor]]:
         return {}

src/powerapi/database/csvdb.py

@@ -26,13 +26,13 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
 import csv
 import os
 
-from typing import List, Type
-from powerapi.report.report import Report, CSV_HEADER_COMMON
 from powerapi.database.base_db import BaseDB, IterDB
 from powerapi.exception import PowerAPIException
+from powerapi.report.report import Report, CSV_HEADER_COMMON
 from powerapi.utils import utils
 
 # Array of field that will not be considered as a group
@@ -192,7 +192,7 @@ class CsvDB(BaseDB):
     a CsvDB instance can be define by its current path
     """
 
-    def __init__(self, report_type: Type[Report], tags: List[str], current_path="/tmp/csvdbtest", files=[]):
+    def __init__(self, report_type: type[Report], tags: list[str], current_path="/tmp/csvdbtest", files=[]):
         """
         :param current_path: Current path where read/write files
         """
@@ -294,7 +294,7 @@ def save(self, report: Report):
                 for row in values:
                     writer.writerow(row)
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
         Allow to save a batch of report
 

src/powerapi/database/file_db.py

@@ -27,11 +27,11 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-from datetime import datetime
+import json
 import logging
-from typing import List, Type
 import os
-import json
+from datetime import datetime
+
 from powerapi.database.base_db import BaseDB, DBError, IterDB
 from powerapi.report import Report
 
@@ -93,7 +93,7 @@ class FileDB(BaseDB):
     Allow to handle a FileDB database in reading or writing.
     """
 
-    def __init__(self, report_type: Type[Report], filename: str):
+    def __init__(self, report_type: type[Report], filename: str):
         """
         :param report_type:        Type of the report handled by this database
         :param filename:        Name of the file containing the report
@@ -142,7 +142,7 @@ def save(self, report: Report):
         with open(self.filename, 'w', encoding='utf-8') as file_object:
             file_object.write(str(final_dict))
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
         Allow to save a batch of data
 

src/powerapi/database/influxdb2.py

@@ -28,7 +28,6 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 import logging
-from typing import List, Type
 from urllib.parse import urlparse
 try:
     from influxdb_client import InfluxDBClient, WriteOptions
@@ -54,7 +53,7 @@ class InfluxDB2(BaseDB):
         Allow to handle a InfluxDB database in reading or writing.
     """
 
-    def __init__(self, report_type: Type[Report], url: str, org: str, bucket_name: str, token: str, tags: List[str],
+    def __init__(self, report_type: type[Report], url: str, org: str, bucket_name: str, token: str, tags: list[str],
                  port=None):
         """
             :param report_type:     Type of the report handled by this database
@@ -140,7 +139,7 @@ def save(self, report: Report):
         """
         self.save_many([report])
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
             Save a batch of data
 

src/powerapi/database/mongodb.py

@@ -26,14 +26,14 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
 import logging
 try:
     import pymongo
     import pymongo.errors
 except ImportError:
     logging.getLogger().info("PyMongo is not installed.")
 
-from typing import List, Type
 from powerapi.database.base_db import BaseDB, DBError, IterDB
 from powerapi.report import Report
 
@@ -93,7 +93,7 @@ class MongoDB(BaseDB):
     Allow to handle a MongoDB database in reading or writing.
     """
 
-    def __init__(self, report_type: Type[Report], uri: str, db_name: str, collection_name: str):
+    def __init__(self, report_type: type[Report], uri: str, db_name: str, collection_name: str):
         """
         :param report_type:        Type of the report handled by this database
         :param uri:             URI of the MongoDB server
@@ -164,7 +164,7 @@ def save(self, report: Report):
         """
         self.collection.insert_one(self.report_type.to_mongodb(report))
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
         Allow to save a batch of data
 

src/powerapi/database/opentsdb.py

@@ -26,14 +26,13 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
 import logging
 try:
     from opentsdb import TSDBClient
 except ImportError:
     logging.getLogger().info("opentsdb-py is not installed.")
 
-from typing import List, Type
-
 from powerapi.report import PowerReport, Report
 from .base_db import BaseDB, DBError
 
@@ -51,7 +50,7 @@ class OpenTSDB(BaseDB):
     Allow to handle an OpenTSDB database to save PowerReport.
     """
 
-    def __init__(self, report_type: Type[Report], host: str, port, metric_name: str):
+    def __init__(self, report_type: type[Report], host: str, port, metric_name: str):
         """
         :param host:             host of the OpenTSDB server
         :param port:            port of the OpenTSDB server
@@ -105,7 +104,7 @@ def save(self, report: PowerReport):
         self.client.send(self.metric_name, report.power, timestamp=int(report.timestamp.timestamp()),
                          host=report.target)
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
         Save a batch of data
 

src/powerapi/database/prometheus_db.py

@@ -28,7 +28,6 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 import logging
-from typing import List, Type
 
 try:
     from prometheus_client import start_http_server, Gauge
@@ -55,8 +54,8 @@ class BasePrometheusDB(BaseDB):
     Base class to expose data to prometheus instance
     """
 
-    def __init__(self, report_type: Type[Report], port: int, metric_name: str,
-                 tags: List[str], metric_description: str = DEFAULT_METRIC_DESCRIPTION, address: str = DEFAULT_ADDRESS):
+    def __init__(self, report_type: type[Report], port: int, metric_name: str,
+                 tags: list[str], metric_description: str = DEFAULT_METRIC_DESCRIPTION, address: str = DEFAULT_ADDRESS):
         BaseDB.__init__(self, report_type)
         self.address = address
         self.port = port
@@ -85,8 +84,8 @@ class PrometheusDB(BasePrometheusDB):
     It can only be used with a pusher actor
     """
 
-    def __init__(self, report_type: Type[Report], port: int, address: str, metric_name: str, metric_description: str,
-                 tags: List[str]):
+    def __init__(self, report_type: type[Report], port: int, address: str, metric_name: str, metric_description: str,
+                 tags: list[str]):
         """
         :param address: address that exposes the metric
         :param port: port used to expose the metric
@@ -176,7 +175,7 @@ def save(self, report: Report):
 
         self._expose_data(key, measure)
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         """
         Save a batch of data
 

src/powerapi/database/socket_db.py

@@ -32,7 +32,7 @@
 from queue import SimpleQueue, Empty
 from socketserver import ThreadingMixIn, TCPServer, StreamRequestHandler
 from threading import Thread
-from typing import Type, Iterator
+from typing import Iterator
 
 from powerapi.database.base_db import IterDB, BaseDB
 from powerapi.report import Report
@@ -135,7 +135,7 @@ class SocketDB(BaseDB):
     Database implementation that exposes a TCP socket the clients can connect to.
     """
 
-    def __init__(self, report_type: Type[Report], host: str, port: int):
+    def __init__(self, report_type: type[Report], host: str, port: int):
         """
         :param report_type: The type of report to create
         :param host: The host address to listen on

src/powerapi/database/virtiofs_db.py

@@ -27,10 +27,8 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-import re
 import os
-
-from typing import List, Type
+import re
 
 from powerapi.report import Report
 from .base_db import BaseDB, DBError
@@ -52,7 +50,7 @@ class VirtioFSDB(BaseDB):
 
     A regular expression must be given by the VM manager to extract vm name from target name. VM name is used to find directory that contains the output file.
     """
-    def __init__(self, report_type: Type[Report], vm_name_regexp: str, root_directory_name: str,
+    def __init__(self, report_type: type[Report], vm_name_regexp: str, root_directory_name: str,
                  vm_directory_name_prefix: str = '', vm_directory_name_suffix: str = ''):
         """
         :param vm_name_regexp:           regexp used to extract vm name from report. The regexp must match the name of the target
@@ -97,6 +95,6 @@ def save(self, report: Report):
         with open(vm_filename_path + vm_filename, 'w', encoding='utf-8') as vm_file:
             vm_file.write(str(power))
 
-    def save_many(self, reports: List[Report]):
+    def save_many(self, reports: list[Report]):
         for report in reports:
             self.save(report)