From 7a55903b014206acf3492dac9defcc84ad8bf365 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mose=20M=C3=BCller?= Date: Tue, 19 Sep 2023 14:12:43 +0200 Subject: [PATCH] removing legacy type hints ( Any: +def process_callable_attribute(attr: Any, args: dict[str, Any]) -> Any: converted_args_or_error_msg = convert_arguments_to_hinted_types( args, get_type_hints(attr) ) @@ -101,7 +94,7 @@ class DataService(rpyc.Service, AbstractDataService): attr: Any, value: Any, index: Optional[int], - path_list: List[str], + path_list: list[str], ) -> None: if isinstance(attr, Enum): update_value_if_changed(target_obj, attr_name, attr.__class__[value]) @@ -142,7 +135,7 @@ class DataService(rpyc.Service, AbstractDataService): with open(self._filename, "r") as f: # Load JSON data from file and update class attributes with these # values - self.load_DataService_from_JSON(cast(Dict[str, Any], json.load(f))) + self.load_DataService_from_JSON(cast(dict[str, Any], json.load(f))) def write_to_file(self) -> None: """ @@ -160,7 +153,7 @@ class DataService(rpyc.Service, AbstractDataService): 'Skipping "write_to_file"...' ) - def load_DataService_from_JSON(self, json_dict: Dict[str, Any]) -> None: + def load_DataService_from_JSON(self, json_dict: dict[str, Any]) -> None: # Traverse the serialized representation and set the attributes of the class serialized_class = self.serialize() for path in generate_paths_from_DataService_dict(json_dict): @@ -196,7 +189,7 @@ class DataService(rpyc.Service, AbstractDataService): f'"{class_value_type}". Ignoring value from JSON file...' ) - def serialize(self) -> Dict[str, Dict[str, Any]]: # noqa + def serialize(self) -> dict[str, dict[str, Any]]: # noqa """ Serializes the instance into a dictionary, preserving the structure of the instance. @@ -225,7 +218,7 @@ class DataService(rpyc.Service, AbstractDataService): Returns: dict: The serialized instance. """ - result: Dict[str, Dict[str, Any]] = {} + result: dict[str, dict[str, Any]] = {} # Get the dictionary of the base class base_set = set(type(super()).__dict__) @@ -302,7 +295,7 @@ class DataService(rpyc.Service, AbstractDataService): sig = inspect.signature(value) # Store parameters and their anotations in a dictionary - parameters: Dict[str, Optional[str]] = {} + parameters: dict[str, Optional[str]] = {} for k, v in sig.parameters.items(): annotation = v.annotation if annotation is not inspect._empty: @@ -363,7 +356,7 @@ class DataService(rpyc.Service, AbstractDataService): def update_DataService_attribute( self, - path_list: List[str], + path_list: list[str], attr_name: str, value: Any, ) -> None: diff --git a/src/pydase/data_service/data_service_list.py b/src/pydase/data_service/data_service_list.py index 0df3723..670de88 100644 --- a/src/pydase/data_service/data_service_list.py +++ b/src/pydase/data_service/data_service_list.py @@ -1,12 +1,4 @@ -import sys - -if sys.version_info < (3, 9): - from typing import Callable, List # noqa -else: - from collections.abc import Callable - - List = list - +from collections.abc import Callable from typing import Any, Union from pydase.utils.warnings import ( @@ -38,11 +30,11 @@ class DataServiceList(list): def __init__( self, - *args: List[Any], - callback: Union[List[Callable[[int, Any], None]], None] = None, + *args: list[Any], + callback: Union[list[Callable[[int, Any], None]], None] = None, **kwargs: Any, ) -> None: - self.callbacks: List[Callable[[int, Any], None]] = [] + self.callbacks: list[Callable[[int, Any], None]] = [] if isinstance(callback, list): self.callbacks = callback diff --git a/src/pydase/data_service/task_manager.py b/src/pydase/data_service/task_manager.py index 9a204fc..92e227e 100644 --- a/src/pydase/data_service/task_manager.py +++ b/src/pydase/data_service/task_manager.py @@ -2,13 +2,7 @@ from __future__ import annotations import asyncio import inspect -import sys - -if sys.version_info < (3, 9): - from typing import Callable # noqa -else: - from collections.abc import Callable - +from collections.abc import Callable from functools import wraps from typing import TYPE_CHECKING, Any, TypedDict, Union diff --git a/src/pydase/server/server.py b/src/pydase/server/server.py index ec0de83..75fcee8 100644 --- a/src/pydase/server/server.py +++ b/src/pydase/server/server.py @@ -1,20 +1,12 @@ import asyncio import os import signal -import sys import threading from concurrent.futures import ThreadPoolExecutor from enum import Enum from types import FrameType from typing import Any, Optional, Protocol, TypedDict, Union -if sys.version_info < (3, 9): - from typing import Dict, List, Type # noqa -else: - Dict = dict - List = list - Type = type - import uvicorn from loguru import logger from rpyc import ( @@ -87,9 +79,9 @@ class AdditionalServer(TypedDict): it's instantiated. """ - server: Type[AdditionalServerProtocol] + server: type[AdditionalServerProtocol] port: int - kwargs: Dict[str, Any] + kwargs: dict[str, Any] class Server: @@ -171,8 +163,8 @@ class Server: enable_rpc: bool = True, enable_web: bool = True, use_forking_server: bool = False, - web_settings: Dict[str, Any] = {}, - additional_servers: List[AdditionalServer] = [], + web_settings: dict[str, Any] = {}, + additional_servers: list[AdditionalServer] = [], **kwargs: Any, ) -> None: self._service = service @@ -187,9 +179,9 @@ class Server: self._rpc_server_type = ForkingServer if use_forking_server else ThreadedServer self._additional_servers = additional_servers self.should_exit = False - self.servers: Dict[str, asyncio.Future[Any]] = {} + self.servers: dict[str, asyncio.Future[Any]] = {} self.executor: Union[ThreadPoolExecutor, None] = None - self._info: Dict[str, Any] = { + self._info: dict[str, Any] = { "name": self._service.get_service_name(), "version": __version__, "rpc_port": self._rpc_port, @@ -394,7 +386,7 @@ class Server: self.should_exit = True def custom_exception_handler( - self, loop: asyncio.AbstractEventLoop, context: Dict[str, Any] + self, loop: asyncio.AbstractEventLoop, context: dict[str, Any] ) -> None: # if any background task creates an unhandled exception, shut down the entire # loop. It's possible we don't want to do this, maybe make this optional in the diff --git a/src/pydase/server/web_server.py b/src/pydase/server/web_server.py index 808f778..2b508dd 100644 --- a/src/pydase/server/web_server.py +++ b/src/pydase/server/web_server.py @@ -1,12 +1,6 @@ -import sys from pathlib import Path from typing import Any, TypedDict, Union -if sys.version_info < (3, 9): - from typing import Dict # noqa -else: - Dict = dict - import socketio from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -56,7 +50,7 @@ class WebAPI: frontend: Union[str, Path, None] = None, css: Union[str, Path, None] = None, enable_CORS: bool = True, - info: Dict[str, Any] = {}, + info: dict[str, Any] = {}, *args: Any, **kwargs: Any, ): @@ -113,11 +107,11 @@ class WebAPI: return self.service.get_service_name() @app.get("/info") - def info() -> Dict[str, Any]: + def info() -> dict[str, Any]: return self.info @app.get("/service-properties") - def service_properties() -> Dict[str, Any]: + def service_properties() -> dict[str, Any]: return self.service.serialize() app.mount( diff --git a/src/pydase/utils/helpers.py b/src/pydase/utils/helpers.py index f3c1ea0..d946eca 100644 --- a/src/pydase/utils/helpers.py +++ b/src/pydase/utils/helpers.py @@ -1,21 +1,13 @@ import re -import sys from itertools import chain from typing import Any, Optional, Union, cast -if sys.version_info < (3, 9): - from typing import Dict, List, Tuple # noqa -else: - Dict = dict - List = list - Tuple = tuple - from loguru import logger STANDARD_TYPES = ("int", "float", "bool", "str", "Enum", "NoneType", "Quantity") -def get_class_and_instance_attributes(obj: Any) -> Dict[str, Any]: +def get_class_and_instance_attributes(obj: Any) -> dict[str, Any]: """Dictionary containing all attributes (both instance and class level) of a given object. @@ -30,7 +22,7 @@ def get_class_and_instance_attributes(obj: Any) -> Dict[str, Any]: return attrs -def get_object_attr_from_path(target_obj: Any, path: List[str]) -> Any: +def get_object_attr_from_path(target_obj: Any, path: list[str]) -> Any: """ Traverse the object tree according to the given path. @@ -65,7 +57,7 @@ def get_object_attr_from_path(target_obj: Any, path: List[str]) -> Any: def generate_paths_from_DataService_dict( data: dict, parent_path: str = "" -) -> List[str]: +) -> list[str]: """ Recursively generate paths from a dictionary representing a DataService object. @@ -135,8 +127,8 @@ def generate_paths_from_DataService_dict( def extract_dict_or_list_entry( - data: Dict[str, Any], key: str -) -> Union[Dict[str, Any], None]: + data: dict[str, Any], key: str +) -> Union[dict[str, Any], None]: """ Extract a nested dictionary or list entry based on the provided key. @@ -188,7 +180,7 @@ def extract_dict_or_list_entry( else: logger.error(f"Invalid index format in key: {key}") - current_data: Union[Dict[str, Any], List[Dict[str, Any]], None] = data.get( + current_data: Union[dict[str, Any], list[dict[str, Any]], None] = data.get( attr_name, None ) if not isinstance(current_data, dict): @@ -207,14 +199,14 @@ def extract_dict_or_list_entry( # When the attribute is a class instance, the attributes are nested in the # "value" key if current_data["type"] not in STANDARD_TYPES: - current_data = cast(Dict[str, Any], current_data.get("value", None)) # type: ignore + current_data = cast(dict[str, Any], current_data.get("value", None)) # type: ignore assert isinstance(current_data, dict) return current_data def get_nested_value_from_DataService_by_path_and_key( - data: Dict[str, Any], path: str, key: str = "value" + data: dict[str, Any], path: str, key: str = "value" ) -> Any: """ Get the value associated with a specific key from a dictionary given a path. @@ -260,8 +252,8 @@ def get_nested_value_from_DataService_by_path_and_key( """ # Split the path into parts - parts: List[str] = re.split(r"\.", path) # Split by '.' - current_data: Union[Dict[str, Any], None] = data + parts: list[str] = re.split(r"\.", path) # Split by '.' + current_data: Union[dict[str, Any], None] = data for part in parts: if current_data is None: @@ -273,8 +265,8 @@ def get_nested_value_from_DataService_by_path_and_key( def convert_arguments_to_hinted_types( - args: Dict[str, Any], type_hints: Dict[str, Any] -) -> Union[Dict[str, Any], str]: + args: dict[str, Any], type_hints: dict[str, Any] +) -> Union[dict[str, Any], str]: """ Convert the given arguments to their types hinted in the type_hints dictionary. @@ -352,7 +344,7 @@ def update_value_if_changed( logger.error(f"Incompatible arguments: {target}, {attr_name_or_index}.") -def parse_list_attr_and_index(attr_string: str) -> Tuple[str, Optional[int]]: +def parse_list_attr_and_index(attr_string: str) -> tuple[str, Optional[int]]: """ Parses an attribute string and extracts a potential list attribute name and its index. @@ -391,7 +383,7 @@ def parse_list_attr_and_index(attr_string: str) -> Tuple[str, Optional[int]]: return attr_name, index -def get_component_class_names() -> List[str]: +def get_component_class_names() -> list[str]: """ Returns the names of the component classes in a list. diff --git a/tests/__init__.py b/tests/__init__.py index 1cba260..81bd255 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,10 +1,4 @@ -import sys - -if sys.version_info < (3, 10): - from typing import Generator -else: - from collections.abc import Generator - +from collections.abc import Generator from typing import Any import pytest