mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-20 00:10:03 +02:00
removing legacy type hints (<v3.9)
This commit is contained in:
parent
dc432a1238
commit
7a55903b01
@ -1,15 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from abc import ABC
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if sys.version_info < (3, 9):
|
||||
from typing import Dict # noqa
|
||||
else:
|
||||
Dict = dict
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .callback_manager import CallbackManager
|
||||
from .data_service import DataService
|
||||
@ -20,4 +13,4 @@ class AbstractDataService(ABC):
|
||||
__root__: DataService
|
||||
_task_manager: TaskManager
|
||||
_callback_manager: CallbackManager
|
||||
_autostart_tasks: Dict[str, tuple[Any]]
|
||||
_autostart_tasks: dict[str, tuple[Any]]
|
||||
|
@ -1,15 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import sys
|
||||
|
||||
if sys.version_info < (3, 9):
|
||||
from typing import Callable, Dict # noqa
|
||||
else:
|
||||
from collections.abc import Callable
|
||||
|
||||
Dict = dict
|
||||
|
||||
from collections.abc import Callable
|
||||
from typing import TYPE_CHECKING, Any, Union
|
||||
|
||||
from loguru import logger
|
||||
@ -45,7 +37,7 @@ class CallbackManager:
|
||||
This implementation follows the observer pattern, with the DataService instance as
|
||||
the "subject" and the callback functions as the "observers".
|
||||
"""
|
||||
_list_mapping: Dict[int, DataServiceList] = {}
|
||||
_list_mapping: dict[int, DataServiceList] = {}
|
||||
"""
|
||||
A dictionary mapping the id of the original lists to the corresponding
|
||||
DataServiceList instances.
|
||||
|
@ -2,16 +2,9 @@ import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from enum import Enum
|
||||
from typing import Any, Optional, cast, get_type_hints
|
||||
|
||||
if sys.version_info < (3, 9):
|
||||
from typing import Dict, List # noqa
|
||||
else:
|
||||
Dict = dict
|
||||
List = list
|
||||
|
||||
import rpyc
|
||||
from loguru import logger
|
||||
|
||||
@ -35,7 +28,7 @@ from pydase.utils.warnings import (
|
||||
)
|
||||
|
||||
|
||||
def process_callable_attribute(attr: Any, args: Dict[str, Any]) -> 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:
|
||||
|
@ -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
|
||||
|
||||
|
@ -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
|
||||
|
||||
|
@ -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
|
||||
|
@ -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(
|
||||
|
@ -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.
|
||||
|
||||
|
@ -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
|
||||
|
Loading…
x
Reference in New Issue
Block a user