removing python 3.8 from workflows

.flake8

@@ -1,8 +1,8 @@
 [flake8]
-ignore = E501,W503,FS003,F403,F405,E203
+ignore = E501,W503,FS003,F403,F405,E203,UNT001
 include = src
 max-line-length = 88
 max-doc-length = 88
 max-complexity = 7
 max-expression-complexity = 5.5
-use_class_attributes_order_strict_mode=True
+use_class_attributes_order_strict_mode=True

.github/workflows/python-package.yml

@@ -5,37 +5,36 @@ name: Python package
 
 on:
   push:
-    branches: [ "main" ]
+    branches: ['main']
   pull_request:
-    branches: [ "main" ]
+    branches: ['main']
 
 jobs:
   build:
-
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11"]
+        python-version: ['3.9', '3.10', '3.11']
 
     steps:
-    - uses: actions/checkout@v3
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
-      with:
-        python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        python -m pip install poetry
-        poetry install
-    - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        poetry run flake8 src/pydase --count --show-source --statistics
-    - name: Test with pytest
-      run: |
-        poetry run pytest
-    - name: Test with pyright
-      run: |
-        poetry run pyright src/pydase
+      - uses: actions/checkout@v3
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v3
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install poetry
+          poetry install
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          poetry run flake8 src/pydase --count --show-source --statistics
+      - name: Test with pytest
+        run: |
+          poetry run pytest
+      - name: Test with pyright
+        run: |
+          poetry run pyright src/pydase

poetry.toml

@@ -1,2 +1,3 @@
 [virtualenvs]
 in-project = true
+prefer-active-python = true

pyproject.toml

@@ -8,7 +8,7 @@ packages = [{ include = "pydase", from = "src" }]
 
 
 [tool.poetry.dependencies]
-python = "^3.10"
+python = "^3.9"
 rpyc = "^5.3.1"
 loguru = "^0.7.0"
 fastapi = "^0.100.0"

src/pydase/components/image.py

@@ -1,7 +1,7 @@
 import base64
 import io
 from pathlib import Path
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Union
 from urllib.request import urlopen
 
 import PIL.Image
@@ -29,7 +29,7 @@ class Image(DataService):
     def format(self) -> str:
         return self._format
 
-    def load_from_path(self, path: Path | str) -> None:
+    def load_from_path(self, path: Union[Path, str]) -> None:
         with PIL.Image.open(path) as image:
             self._load_from_PIL(image)
 
@@ -68,7 +68,7 @@ class Image(DataService):
         else:
             logger.error("Image format is 'None'. Skipping...")
 
-    def _get_image_format_from_bytes(self, value_: bytes) -> str | None:
+    def _get_image_format_from_bytes(self, value_: bytes) -> Union[str, None]:
         image_data = base64.b64decode(value_)
         # Create a writable memory buffer for the image
         image_buffer = io.BytesIO(image_data)

src/pydase/components/number_slider.py

@@ -1,4 +1,4 @@
-from typing import Any, Literal
+from typing import Any, Literal, Union
 
 from loguru import logger
 
@@ -39,11 +39,11 @@ class NumberSlider(DataService):
 
     def __init__(
         self,
-        value: float | int = 0,
+        value: Union[float, int] = 0,
         min: float = 0.0,
         max: float = 100.0,
-        step_size: float | int = 1.0,
-        type: Literal["int"] | Literal["float"] = "float",
+        step_size: Union[float, int] = 1.0,
+        type: Union[Literal["int"], Literal["float"]] = "float",
     ) -> None:
         if type not in {"float", "int"}:
             logger.error(f"Unknown type '{type}'. Using 'float'.")

src/pydase/config.py

@@ -1,9 +1,9 @@
-from typing import Literal
+from typing import Literal, Union
 
 from confz import BaseConfig, EnvSource
 
 
 class OperationMode(BaseConfig):  # type: ignore
-    environment: Literal["development"] | Literal["production"] = "development"
+    environment: Union[Literal["development"], Literal["production"]] = "development"
 
     CONFIG_SOURCES = EnvSource(allow=["ENVIRONMENT"])

src/pydase/data_service/callback_manager.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import inspect
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Union
 
 from loguru import logger
 
@@ -206,8 +206,8 @@ class CallbackManager:
 
     def __register_recursive_parameter_callback(
         self,
-        obj: "AbstractDataService | DataServiceList",
-        callback: Callable[[str | int, Any], None],
+        obj: Union["AbstractDataService", DataServiceList],
+        callback: Callable[[Union[str, int], Any], None],
     ) -> None:
         """
         Register callback to a DataService or DataServiceList instance and its nested
@@ -222,7 +222,7 @@ class CallbackManager:
         if isinstance(obj, DataServiceList):
             # emits callback when item in list gets reassigned
             obj.add_callback(callback=callback)
-            obj_list: DataServiceList | list[AbstractDataService] = obj
+            obj_list: Union[DataServiceList, list[AbstractDataService]] = obj
         else:
             obj_list = [obj]
 
@@ -337,7 +337,7 @@ class CallbackManager:
 
         # Create and register a callback for the object
         # only emit the notification when the call was registered by the root object
-        callback: Callable[[str, dict[str, Any] | None], None] = (
+        callback: Callable[[str, Union[dict[str, Any], None]], None] = (
             lambda name, status: obj._callback_manager.emit_notification(
                 parent_path=parent_path, name=name, value=status
             )

src/pydase/data_service/data_service_list.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable
-from typing import Any
+from typing import Any, Union
 
 from pydase.utils.warnings import (
     warn_if_instance_class_does_not_inherit_from_DataService,
@@ -31,7 +31,7 @@ class DataServiceList(list):
     def __init__(
         self,
         *args: list[Any],
-        callback: list[Callable[[int, Any], None]] | None = None,
+        callback: Union[list[Callable[[int, Any], None]], None] = None,
         **kwargs: Any,
     ) -> None:
         self.callbacks: list[Callable[[int, Any], None]] = []

src/pydase/data_service/task_manager.py

@@ -4,7 +4,7 @@ import asyncio
 import inspect
 from collections.abc import Callable
 from functools import wraps
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import TYPE_CHECKING, Any, TypedDict, Union
 
 from loguru import logger
 
@@ -82,7 +82,7 @@ class TaskManager:
         """
 
         self.task_status_change_callbacks: list[
-            Callable[[str, dict[str, Any] | None], Any]
+            Callable[[str, Union[dict[str, Any], None]], Any]
         ] = []
         """A list of callback functions to be invoked when the status of a task (start
         or stop) changes."""

src/pydase/server/server.py

@@ -5,7 +5,7 @@ import threading
 from concurrent.futures import ThreadPoolExecutor
 from enum import Enum
 from types import FrameType
-from typing import Any, Optional, Protocol, TypedDict
+from typing import Any, Optional, Protocol, TypedDict, Union
 
 import uvicorn
 from loguru import logger
@@ -180,7 +180,7 @@ class Server:
         self._additional_servers = additional_servers
         self.should_exit = False
         self.servers: dict[str, asyncio.Future[Any]] = {}
-        self.executor: ThreadPoolExecutor | None = None
+        self.executor: Union[ThreadPoolExecutor, None] = None
         self._info: dict[str, Any] = {
             "name": self._service.get_service_name(),
             "version": __version__,

src/pydase/server/web_server.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, TypedDict
+from typing import Any, TypedDict, Union
 
 import socketio
 from fastapi import FastAPI
@@ -47,8 +47,8 @@ class WebAPI:
     def __init__(  # noqa: CFQ002
         self,
         service: DataService,
-        frontend: str | Path | None = None,
-        css: str | Path | None = None,
+        frontend: Union[str, Path, None] = None,
+        css: Union[str, Path, None] = None,
         enable_CORS: bool = True,
         info: dict[str, Any] = {},
         *args: Any,

src/pydase/units.py

@@ -1,21 +1,19 @@
-from typing import TypedDict
+from typing import TypedDict, Union
 
 import pint
+from pint import Quantity
 
-units: pint.UnitRegistry = pint.UnitRegistry()
+units = pint.UnitRegistry()
 units.default_format = "~P"  # pretty and short format
 
-Quantity = pint.Quantity
-Unit = units.Unit
-
 
 class QuantityDict(TypedDict):
-    magnitude: int | float
+    magnitude: Union[int, float]
     unit: str
 
 
 def convert_to_quantity(
-    value: QuantityDict | float | int | Quantity, unit: str = ""
+    value: Union[QuantityDict, float, int, Quantity], unit: str = ""
 ) -> Quantity:
     """
     Convert a given value into a pint.Quantity object with the specified unit.
@@ -47,10 +45,10 @@ def convert_to_quantity(
           will be unitless.
     """
 
-    if isinstance(value, int | float):
-        quantity = float(value) * Unit(unit)
+    if isinstance(value, (int, float)):
+        quantity = float(value) * units(unit)
     elif isinstance(value, dict):
-        quantity = float(value["magnitude"]) * Unit(value["unit"])
+        quantity = float(value["magnitude"]) * units(value["unit"])
     else:
         quantity = value
     return quantity  # type: ignore

src/pydase/utils/helpers.py

@@ -1,13 +1,13 @@
 import re
 from itertools import chain
-from typing import Any, Optional, cast
+from typing import Any, Optional, Union, cast
 
 from loguru import logger
 
 STANDARD_TYPES = ("int", "float", "bool", "str", "Enum", "NoneType", "Quantity")
 
 
-def get_class_and_instance_attributes(obj: object) -> 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.
 
@@ -126,7 +126,9 @@ def generate_paths_from_DataService_dict(
     return paths
 
 
-def extract_dict_or_list_entry(data: dict[str, Any], key: str) -> dict[str, Any] | None:
+def extract_dict_or_list_entry(
+    data: dict[str, Any], key: str
+) -> Union[dict[str, Any], None]:
     """
     Extract a nested dictionary or list entry based on the provided key.
 
@@ -178,7 +180,7 @@ def extract_dict_or_list_entry(data: dict[str, Any], key: str) -> dict[str, Any]
         else:
             logger.error(f"Invalid index format in key: {key}")
 
-    current_data: 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):
@@ -251,7 +253,7 @@ 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: dict[str, Any] | None = data
+    current_data: Union[dict[str, Any], None] = data
 
     for part in parts:
         if current_data is None:
@@ -264,7 +266,7 @@ 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]
-) -> dict[str, Any] | str:
+) -> Union[dict[str, Any], str]:
     """
     Convert the given arguments to their types hinted in the type_hints dictionary.
 
@@ -306,7 +308,7 @@ def convert_arguments_to_hinted_types(
 
 
 def update_value_if_changed(
-    target: Any, attr_name_or_index: str | int, new_value: Any
+    target: Any, attr_name_or_index: Union[str, int], new_value: Any
 ) -> None:
     """
     Updates the value of an attribute or a list element on a target object if the new

src/pydase/utils/logging.py

@@ -1,7 +1,7 @@
 import logging
 import sys
 from types import FrameType
-from typing import Optional
+from typing import Optional, Union
 
 import loguru
 import rpyc
@@ -21,7 +21,7 @@ class InterceptHandler(logging.Handler):
             return
 
         # Get corresponding Loguru level if it exists.
-        level: int | str
+        level: Union[int, str]
         try:
             level = loguru.logger.level(record.levelname).name
         except ValueError: