Merge pull request #172 from tiqi-group/feat/update_client_reconnection

Feat: update client reconnection

README.md

@@ -105,7 +105,7 @@ class Device(pydase.DataService):
 
 if __name__ == "__main__":
     service = Device()
-    pydase.Server(service=service).run()
+    pydase.Server(service=service, web_port=8001).run()
 ```
 
 In the above example, we define a `Device` class that inherits from `pydase.DataService`. 
@@ -122,10 +122,13 @@ import pydase
 
 if __name__ == "__main__":
     service = Device()
-    pydase.Server(service=service).run()
+    pydase.Server(service=service, web_port=8001).run()
 ```
 
-This will start the server, making your `Device` service accessible on [http://localhost:8001](http://localhost:8001).
+This will start the server, making your `Device` service accessible on 
+[http://localhost:8001](http://localhost:8001). The port number for the web server can 
+be customised in the server constructor or through environment variables and defaults 
+to `8001`.
 
 ### Accessing the Web Interface
 
@@ -144,7 +147,7 @@ import pydase
 
 # Replace the hostname and port with the IP address and the port of the machine where 
 # the service is running, respectively
-client_proxy = pydase.Client(url="ws://<ip_addr>:<service_port>").proxy
+client_proxy = pydase.Client(url="ws://<ip_addr>:<web_port>").proxy
 # client_proxy = pydase.Client(url="wss://your-domain.ch").proxy  # if your service uses ssl-encryption
 
 # After the connection, interact with the service attributes as if they were local
@@ -170,7 +173,7 @@ import json
 import requests
 
 response = requests.get(
-    "http://<hostname>:<port>/api/v1/get_value?access_path=<full_access_path>"
+    "http://<hostname>:<web_port>/api/v1/get_value?access_path=<full_access_path>"
 )
 serialized_value = json.loads(response.text)
 ```

docs/dev-guide/api.md

@@ -13,6 +13,12 @@
 ::: pydase.components
     handler: python
 
+::: pydase.task
+    handler: python
+    options:
+      inherited_members: false
+      show_submodules: true
+
 ::: pydase.utils.serialization.serializer
     handler: python
 

docs/user-guide/Tasks.md

@@ -1,20 +1,18 @@
 # Understanding Tasks
 
-In `pydase`, a task is defined as an asynchronous function without arguments contained in a class that inherits from `pydase.DataService`. These tasks usually contain a while loop and are designed to carry out periodic functions.
+In `pydase`, a task is defined as an asynchronous function without arguments that is decorated with the `@task` decorator and contained in a class that inherits from `pydase.DataService`. These tasks usually contain a while loop and are designed to carry out periodic functions. For example, a task might be used to periodically read sensor data, update a database, or perform any other recurring job.
 
-For example, a task might be used to periodically read sensor data, update a database, or perform any other recurring job. One core feature of `pydase` is its ability to automatically generate start and stop functions for these tasks. This allows you to control task execution via both the frontend and python clients, giving you flexible and powerful control over your service's operation.
-
-Another powerful feature of `pydase` is its ability to automatically start tasks upon initialization of the service. By specifying the tasks and their arguments in the `_autostart_tasks` dictionary in your service class's `__init__` method, `pydase` will automatically start these tasks when the server is started. Here's an example:
+`pydase` allows you to control task execution via both the frontend and Python clients and can automatically start tasks upon initialization of the service. By using the `@task` decorator with the `autostart=True` argument in your service class, `pydase` will automatically start these tasks when the server is started. Here's an example:
 
 ```python
 import pydase
+from pydase.task.decorator import task
 
 
 class SensorService(pydase.DataService):
     def __init__(self):
         super().__init__()
         self.readout_frequency = 1.0
-        self._autostart_tasks["read_sensor_data"] = ()
 
     def _process_data(self, data: ...) -> None:
         ...
@@ -22,6 +20,7 @@ class SensorService(pydase.DataService):
     def _read_from_sensor(self) -> Any:
         ...
 
+    @task(autostart=True)
     async def read_sensor_data(self):
         while True:
             data = self._read_from_sensor()
@@ -34,6 +33,6 @@ if __name__ == "__main__":
     pydase.Server(service=service).run()
 ```
 
-In this example, `read_sensor_data` is a task that continuously reads data from a sensor. By adding it to the `_autostart_tasks` dictionary, it will automatically start running when `pydase.Server(service).run()` is executed.
-As with all tasks, `pydase` will generate `start_read_sensor_data` and `stop_read_sensor_data` methods, which can be called to manually start and stop the data reading task. The readout frequency can be updated using the `readout_frequency` attribute.
+In this example, `read_sensor_data` is a task that continuously reads data from a sensor. By decorating it with `@task(autostart=True)`, it will automatically start running when `pydase.Server(service).run()` is executed.
 
+The `@task` decorator replaces the function with a task object that has `start()` and `stop()` methods. This means you can control the task execution directly using these methods. For instance, you can manually start or stop the task by calling `service.read_sensor_data.start()` and `service.read_sensor_data.stop()`, respectively.

docs/user-guide/interaction/Python Client.md

@@ -1,60 +1,86 @@
 # Python RPC Client
 
-You can connect to the service using the `pydase.Client`. Below is an example of how to establish a connection to a service and interact with it:
+The [`pydase.Client`][pydase.Client] allows you to connect to a remote `pydase` service using socket.io, facilitating interaction with the service as though it were running locally.
+
+## Basic Usage
 
 ```python
 import pydase
 
-# Replace the hostname and port with the IP address and the port of the machine where 
-# the service is running, respectively
+# Replace <ip_addr> and <service_port> with the appropriate values for your service
 client_proxy = pydase.Client(url="ws://<ip_addr>:<service_port>").proxy
-# client_proxy = pydase.Client(url="wss://your-domain.ch").proxy  # if your service uses ssl-encryption
+# For SSL-encrypted services, use the wss protocol
+# client_proxy = pydase.Client(url="wss://your-domain.ch").proxy
 
 # Interact with the service attributes as if they were local
 client_proxy.voltage = 5.0
 print(client_proxy.voltage)  # Expected output: 5.0
 ```
 
-This example demonstrates setting and retrieving the `voltage` attribute through the client proxy.
-The proxy acts as a local representative of the remote service, enabling straightforward interaction.
+This example shows how to set and retrieve the `voltage` attribute through the client proxy.
+The proxy acts as a local representation of the remote service, enabling intuitive interaction.
 
-The proxy class dynamically synchronizes with the server's exposed attributes. This synchronization allows the proxy to be automatically updated with any attributes or methods that the server exposes, essentially mirroring the server's API. This dynamic updating enables users to interact with the remote service as if they were working with a local object.
+The proxy class automatically synchronizes with the server's attributes and methods, keeping itself up-to-date with any changes. This dynamic synchronization essentially mirrors the server's API, making it feel like you're working with a local object.
 
-## Context Manager
+## Context Manager Support
 
-You can also use the client as a context manager which automatically opens and closes the connection again:
+You can also use the client within a context manager, which automatically handles connection management (i.e., opening and closing the connection):
 
 ```python
 import pydase
 
 
 with pydase.Client(url="ws://localhost:8001") as client:
-    client.proxy.<my_method>()
+    client.proxy.my_method()
 ```
 
+Using the context manager ensures that connections are cleanly closed once the block of code finishes executing.
 
 ## Tab Completion Support
 
-In interactive environments such as Python interpreters and Jupyter notebooks, the proxy class supports tab completion, which allows users to explore available methods and attributes.
+In interactive environments like Python interpreters or Jupyter notebooks, the proxy supports tab completion. This allows users to explore available methods and attributes.
 
-## Integration within Other Services
+## Integrating the Client into Another Service
 
-You can also integrate a client proxy within another service. Here's how you can set it up:
+You can integrate a `pydase` client proxy within another service. Here's an example of how to set this up:
 
 ```python
 import pydase
 
 class MyService(pydase.DataService):
-    # Initialize the client without blocking the constructor
-    proxy = pydase.Client(url="ws://<ip_addr>:<service_port>", block_until_connected=False).proxy
-    # proxy = pydase.Client(url="wss://your-domain.ch", block_until_connected=False).proxy  # communicating with ssl-encrypted service
+    proxy = pydase.Client(
+        url="ws://<ip_addr>:<service_port>",
+        block_until_connected=False
+    ).proxy
+    # For SSL-encrypted services, use the wss protocol
+    # proxy = pydase.Client(
+    #     url="wss://your-domain.ch",
+    #     block_until_connected=False
+    # ).proxy
 
 if __name__ == "__main__":
     service = MyService()
-    # Create a server that exposes this service; adjust the web_port as needed
-    server = pydase.Server(service, web_port=8002). run()
+    # Create a server that exposes this service
+    server = pydase.Server(service, web_port=8002).run()
 ```
 
-In this setup, the `MyService` class has a `proxy` attribute that connects to a `pydase` service located at `<ip_addr>:8001`.
-The `block_until_connected=False` argument allows the service to start up even if the initial connection attempt fails.
-This configuration is particularly useful in distributed systems where services may start in any order.
+In this example:
+- The `MyService` class has a `proxy` attribute that connects to a `pydase` service at `<ip_addr>:<service_port>`.
+- By setting `block_until_connected=False`, the service can start without waiting for the connection to succeed, which is particularly useful in distributed systems where services may initialize in any order.
+
+## Custom `socketio.AsyncClient` Connection Parameters
+
+You can also configure advanced connection options by passing additional arguments to the underlying [`AsyncClient`][socketio.AsyncClient] via `sio_client_kwargs`. This allows you to fine-tune reconnection behaviour, delays, and other settings:
+
+```python
+client = pydase.Client(
+    url="ws://localhost:8001",
+    sio_client_kwargs={
+        "reconnection_attempts": 3,
+        "reconnection_delay": 2,
+        "reconnection_delay_max": 10,
+    }
+).proxy
+```
+
+In this setup, the client will attempt to reconnect three times, with an initial delay of 2 seconds (each successive attempt doubles this delay) and a maximum delay of 10 seconds between attempts.

frontend/src/components/GenericComponent.tsx

@@ -4,7 +4,6 @@ import { NumberComponent, NumberObject } from "./NumberComponent";
 import { SliderComponent } from "./SliderComponent";
 import { EnumComponent } from "./EnumComponent";
 import { MethodComponent } from "./MethodComponent";
-import { AsyncMethodComponent } from "./AsyncMethodComponent";
 import { StringComponent } from "./StringComponent";
 import { ListComponent } from "./ListComponent";
 import { DataServiceComponent, DataServiceJSON } from "./DataServiceComponent";
@@ -17,6 +16,7 @@ import { updateValue } from "../socket";
 import { DictComponent } from "./DictComponent";
 import { parseFullAccessPath } from "../utils/stateUtils";
 import { SerializedEnum, SerializedObject } from "../types/SerializedObject";
+import { TaskComponent, TaskStatus } from "./TaskComponent";
 
 interface GenericComponentProps {
   attribute: SerializedObject;
@@ -144,30 +144,16 @@ export const GenericComponent = React.memo(
         />
       );
     } else if (attribute.type === "method") {
-      if (!attribute.async) {
-        return (
-          <MethodComponent
-            fullAccessPath={fullAccessPath}
-            docString={attribute.doc}
-            addNotification={addNotification}
-            displayName={displayName}
-            id={id}
-            render={attribute.frontend_render}
-          />
-        );
-      } else {
-        return (
-          <AsyncMethodComponent
-            fullAccessPath={fullAccessPath}
-            docString={attribute.doc}
-            value={attribute.value as "RUNNING" | null}
-            addNotification={addNotification}
-            displayName={displayName}
-            id={id}
-            render={attribute.frontend_render}
-          />
-        );
-      }
+      return (
+        <MethodComponent
+          fullAccessPath={fullAccessPath}
+          docString={attribute.doc}
+          addNotification={addNotification}
+          displayName={displayName}
+          id={id}
+          render={attribute.frontend_render}
+        />
+      );
     } else if (attribute.type === "str") {
       return (
         <StringComponent
@@ -182,6 +168,17 @@ export const GenericComponent = React.memo(
           id={id}
         />
       );
+    } else if (attribute.type == "Task") {
+      return (
+        <TaskComponent
+          fullAccessPath={fullAccessPath}
+          docString={attribute.doc}
+          status={attribute.value["status"].value as TaskStatus}
+          addNotification={addNotification}
+          displayName={displayName}
+          id={id}
+        />
+      );
     } else if (attribute.type === "DataService") {
       return (
         <DataServiceComponent

frontend/src/components/NumberComponent.tsx

@@ -132,6 +132,8 @@ const handleNumericKey = (
   selectionStart: number,
   selectionEnd: number,
 ) => {
+  let newValue = value;
+
   // Check if a number key or a decimal point key is pressed
   if (key === "." && value.includes(".")) {
     // Check if value already contains a decimal. If so, ignore input.
@@ -139,14 +141,34 @@ const handleNumericKey = (
     return { value, selectionStart };
   }
 
-  let newValue = value;
+  // Handle minus sign input
+  if (key === "-") {
+    if (selectionStart === 0 && selectionEnd > selectionStart) {
+      // Replace selection with minus if selection starts at 0
+      newValue = "-" + value.slice(selectionEnd);
+      selectionStart = 1;
+    } else if (selectionStart === 0 && !value.startsWith("-")) {
+      // Add minus at the beginning if it doesn't exist
+      newValue = "-" + value;
+      selectionStart = 1;
+    } else if (
+      (selectionStart === 0 || selectionStart === 1) &&
+      value.startsWith("-")
+    ) {
+      // Remove minus if it exists
+      newValue = value.slice(1);
+      selectionStart = 0;
+    }
+
+    return { value: newValue, selectionStart };
+  }
 
   // Add the new key at the cursor's position
   if (selectionEnd > selectionStart) {
     // If there is a selection, replace it with the key
     newValue = value.slice(0, selectionStart) + key + value.slice(selectionEnd);
   } else {
-    // otherwise, append the key after the selection start
+    // Otherwise, insert the key at the cursor position
     newValue = value.slice(0, selectionStart) + key + value.slice(selectionStart);
   }
 
@@ -201,17 +223,7 @@ export const NumberComponent = React.memo((props: NumberComponentProps) => {
       // Select everything when pressing Ctrl + a
       inputTarget.setSelectionRange(0, value.length);
       return;
-    } else if (key === "-") {
-      if (selectionStart === 0 && !value.startsWith("-")) {
-        newValue = "-" + value;
-        selectionStart++;
-      } else if (value.startsWith("-") && selectionStart === 1) {
-        newValue = value.substring(1); // remove minus sign
-        selectionStart--;
-      } else {
-        return; // Ignore "-" pressed in other positions
-      }
-    } else if (key >= "0" && key <= "9") {
+    } else if ((key >= "0" && key <= "9") || key === "-") {
       // Check if a number key or a decimal point key is pressed
       ({ value: newValue, selectionStart } = handleNumericKey(
         key,

frontend/src/components/TaskComponent.tsx

@@ -5,67 +5,51 @@ import { DocStringComponent } from "./DocStringComponent";
 import { LevelName } from "./NotificationsComponent";
 import useRenderCount from "../hooks/useRenderCount";
 
-interface AsyncMethodProps {
+export type TaskStatus = "RUNNING" | "NOT_RUNNING";
+
+interface TaskProps {
   fullAccessPath: string;
-  value: "RUNNING" | null;
   docString: string | null;
-  hideOutput?: boolean;
+  status: TaskStatus;
   addNotification: (message: string, levelname?: LevelName) => void;
   displayName: string;
   id: string;
-  render: boolean;
 }
 
-export const AsyncMethodComponent = React.memo((props: AsyncMethodProps) => {
-  const {
-    fullAccessPath,
-    docString,
-    value: runningTask,
-    addNotification,
-    displayName,
-    id,
-  } = props;
-
-  // Conditional rendering based on the 'render' prop.
-  if (!props.render) {
-    return null;
-  }
+export const TaskComponent = React.memo((props: TaskProps) => {
+  const { fullAccessPath, docString, status, addNotification, displayName, id } = props;
 
   const renderCount = useRenderCount();
   const formRef = useRef(null);
   const [spinning, setSpinning] = useState(false);
-  const name = fullAccessPath.split(".").at(-1)!;
-  const parentPath = fullAccessPath.slice(0, -(name.length + 1));
 
   useEffect(() => {
     let message: string;
 
-    if (runningTask === null) {
-      message = `${fullAccessPath} task was stopped.`;
-    } else {
+    if (status === "RUNNING") {
       message = `${fullAccessPath} was started.`;
+    } else {
+      message = `${fullAccessPath} was stopped.`;
     }
+
     addNotification(message);
     setSpinning(false);
-  }, [props.value]);
+  }, [status]);
 
   const execute = async (event: React.FormEvent) => {
     event.preventDefault();
-    let method_name: string;
 
-    if (runningTask !== undefined && runningTask !== null) {
-      method_name = `stop_${name}`;
-    } else {
-      method_name = `start_${name}`;
-    }
+    const method_name = status == "RUNNING" ? "stop" : "start";
 
-    const accessPath = [parentPath, method_name].filter((element) => element).join(".");
+    const accessPath = [fullAccessPath, method_name]
+      .filter((element) => element)
+      .join(".");
     setSpinning(true);
     runMethod(accessPath);
   };
 
   return (
-    <div className="component asyncMethodComponent" id={id}>
+    <div className="component taskComponent" id={id}>
       {process.env.NODE_ENV === "development" && <div>Render count: {renderCount}</div>}
       <Form onSubmit={execute} ref={formRef}>
         <InputGroup>
@@ -76,7 +60,7 @@ export const AsyncMethodComponent = React.memo((props: AsyncMethodProps) => {
           <Button id={`button-${id}`} type="submit">
             {spinning ? (
               <Spinner size="sm" role="status" aria-hidden="true" />
-            ) : runningTask === "RUNNING" ? (
+            ) : status === "RUNNING" ? (
               "Stop "
             ) : (
               "Start "
@@ -88,4 +72,4 @@ export const AsyncMethodComponent = React.memo((props: AsyncMethodProps) => {
   );
 });
 
-AsyncMethodComponent.displayName = "AsyncMethodComponent";
+TaskComponent.displayName = "TaskComponent";

frontend/src/types/SerializedObject.ts

@@ -77,7 +77,12 @@ type SerializedException = SerializedObjectBase & {
   type: "Exception";
 };
 
-type DataServiceTypes = "DataService" | "Image" | "NumberSlider" | "DeviceConnection";
+type DataServiceTypes =
+  | "DataService"
+  | "Image"
+  | "NumberSlider"
+  | "DeviceConnection"
+  | "Task";
 
 type SerializedDataService = SerializedObjectBase & {
   name: string;

mkdocs.yml

@@ -54,6 +54,7 @@ plugins:
           - https://docs.python.org/3/objects.inv
           - https://docs.pydantic.dev/latest/objects.inv
           - https://confz.readthedocs.io/en/latest/objects.inv
+          - https://python-socketio.readthedocs.io/en/stable/objects.inv
         options:
           show_source: true
           inherited_members: true

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pydase"
-version = "0.9.1"
+version = "0.10.6"
 description = "A flexible and robust Python library for creating, managing, and interacting with data services, with built-in support for web and RPC servers, and customizable features for diverse use cases."
 authors = ["Mose Mueller <mosmuell@ethz.ch>"]
 readme = "README.md"

src/pydase/client/client.py

@@ -2,12 +2,12 @@ import asyncio
 import logging
 import sys
 import threading
-from typing import TypedDict, cast
+from typing import TYPE_CHECKING, Any, TypedDict, cast
 
 import socketio  # type: ignore
 
-import pydase.components
-from pydase.client.proxy_loader import ProxyClassMixin, ProxyLoader
+from pydase.client.proxy_class import ProxyClass
+from pydase.client.proxy_loader import ProxyLoader
 from pydase.utils.serialization.deserializer import loads
 from pydase.utils.serialization.types import SerializedDataService, SerializedObject
 
@@ -31,50 +31,7 @@ class NotifyDict(TypedDict):
 
 def asyncio_loop_thread(loop: asyncio.AbstractEventLoop) -> None:
     asyncio.set_event_loop(loop)
-    try:
-        loop.run_forever()
-    except RuntimeError:
-        logger.debug("Tried starting even loop, but it is running already")
-
-
-class ProxyClass(ProxyClassMixin, pydase.components.DeviceConnection):
-    """
-    A proxy class that serves as the interface for interacting with device connections
-    via a socket.io client in an asyncio environment.
-
-    Args:
-        sio_client:
-            The socket.io client instance used for asynchronous communication with the
-            pydase service server.
-        loop:
-            The event loop in which the client operations are managed and executed.
-
-    This class is used to create a proxy object that behaves like a local representation
-    of a remote pydase service, facilitating direct interaction as if it were local
-    while actually communicating over network protocols.
-    It can also be used as an attribute of a pydase service itself, e.g.
-
-    ```python
-    import pydase
-
-
-    class MyService(pydase.DataService):
-        proxy = pydase.Client(
-            hostname="...", port=8001, block_until_connected=False
-        ).proxy
-
-
-    if __name__ == "__main__":
-        service = MyService()
-        server = pydase.Server(service, web_port=8002).run()
-    ```
-    """
-
-    def __init__(
-        self, sio_client: socketio.AsyncClient, loop: asyncio.AbstractEventLoop
-    ) -> None:
-        super().__init__()
-        self._initialise(sio_client=sio_client, loop=loop)
+    loop.run_forever()
 
 
 class Client:
@@ -88,15 +45,35 @@ class Client:
         url:
             The URL of the pydase Socket.IO server. This should always contain the
             protocol and the hostname.
-
-            Examples:
-
-            - `wss://my-service.example.com`  # for secure connections, use wss
-            - `ws://localhost:8001`
         block_until_connected:
             If set to True, the constructor will block until the connection to the
             service has been established. This is useful for ensuring the client is
             ready to use immediately after instantiation. Default is True.
+        sio_client_kwargs:
+            Additional keyword arguments passed to the underlying
+            [`AsyncClient`][socketio.AsyncClient]. This allows fine-tuning of the
+            client's behaviour (e.g., reconnection attempts or reconnection delay).
+            Default is an empty dictionary.
+
+    Example:
+        The following example demonstrates a `Client` instance that connects to another
+        pydase service, while customising some of the connection settings for the
+        underlying [`AsyncClient`][socketio.AsyncClient].
+
+        ```python
+        pydase.Client(url="ws://localhost:8001", sio_client_kwargs={
+            "reconnection_attempts": 2,
+            "reconnection_delay": 2,
+            "reconnection_delay_max": 8,
+        })
+        ```
+
+        When connecting to a server over a secure connection (i.e., the server is using
+        SSL/TLS encryption), make sure that the `wss` protocol is used instead of `ws`:
+
+        ```python
+        pydase.Client(url="wss://my-service.example.com")
+        ```
     """
 
     def __init__(
@@ -104,11 +81,14 @@ class Client:
         *,
         url: str,
         block_until_connected: bool = True,
+        sio_client_kwargs: dict[str, Any] = {},
     ):
         self._url = url
-        self._sio = socketio.AsyncClient()
+        self._sio = socketio.AsyncClient(**sio_client_kwargs)
         self._loop = asyncio.new_event_loop()
-        self.proxy = ProxyClass(sio_client=self._sio, loop=self._loop)
+        self.proxy = ProxyClass(
+            sio_client=self._sio, loop=self._loop, reconnect=self.connect
+        )
         """A proxy object representing the remote service, facilitating interaction as
         if it were local."""
         self._thread = threading.Thread(
@@ -164,7 +144,13 @@ class Client:
             self.proxy, serialized_object=serialized_object
         )
         serialized_object["type"] = "DeviceConnection"
-        self.proxy._notify_changed("", loads(serialized_object))
+        if self.proxy._service_representation is not None:
+            # need to use object.__setattr__ to not trigger an observer notification
+            object.__setattr__(self.proxy, "_service_representation", serialized_object)
+
+            if TYPE_CHECKING:
+                self.proxy._service_representation = serialized_object  # type: ignore
+        self.proxy._notify_changed("", self.proxy)
         self.proxy._connected = True
 
     async def _handle_disconnect(self) -> None:

src/pydase/client/proxy_class.py

@@ -0,0 +1,112 @@
+import asyncio
+import logging
+from collections.abc import Callable
+from copy import deepcopy
+from typing import TYPE_CHECKING, cast
+
+import socketio  # type: ignore
+
+import pydase.components
+from pydase.client.proxy_loader import ProxyClassMixin
+from pydase.utils.helpers import get_attribute_doc
+from pydase.utils.serialization.types import SerializedDataService, SerializedObject
+
+logger = logging.getLogger(__name__)
+
+
+class ProxyClass(ProxyClassMixin, pydase.components.DeviceConnection):
+    """
+    A proxy class that serves as the interface for interacting with device connections
+    via a socket.io client in an asyncio environment.
+
+    Args:
+        sio_client:
+            The socket.io client instance used for asynchronous communication with the
+            pydase service server.
+        loop:
+            The event loop in which the client operations are managed and executed.
+        reconnect:
+            The method that is called periodically when the client is not connected.
+
+    This class is used to create a proxy object that behaves like a local representation
+    of a remote pydase service, facilitating direct interaction as if it were local
+    while actually communicating over network protocols.
+    It can also be used as an attribute of a pydase service itself, e.g.
+
+    ```python
+    import pydase
+
+
+    class MyService(pydase.DataService):
+        proxy = pydase.Client(
+            hostname="...", port=8001, block_until_connected=False
+        ).proxy
+
+
+    if __name__ == "__main__":
+        service = MyService()
+        server = pydase.Server(service, web_port=8002).run()
+    ```
+    """
+
+    def __init__(
+        self,
+        sio_client: socketio.AsyncClient,
+        loop: asyncio.AbstractEventLoop,
+        reconnect: Callable[..., None],
+    ) -> None:
+        if TYPE_CHECKING:
+            self._service_representation: None | SerializedObject = None
+
+        super().__init__()
+        pydase.components.DeviceConnection.__init__(self)
+        self._initialise(sio_client=sio_client, loop=loop)
+        object.__setattr__(self, "_service_representation", None)
+        self.reconnect = reconnect
+
+    def serialize(self) -> SerializedObject:
+        if self._service_representation is None:
+            serialization_future = cast(
+                asyncio.Future[SerializedDataService],
+                asyncio.run_coroutine_threadsafe(
+                    self._sio.call("service_serialization"), self._loop
+                ),
+            )
+            # need to use object.__setattr__ to not trigger an observer notification
+            object.__setattr__(
+                self, "_service_representation", serialization_future.result()
+            )
+            if TYPE_CHECKING:
+                self._service_representation = serialization_future.result()
+
+        device_connection_value = cast(
+            dict[str, SerializedObject],
+            pydase.components.DeviceConnection().serialize()["value"],
+        )
+
+        readonly = False
+        doc = get_attribute_doc(self)
+        obj_name = self.__class__.__name__
+
+        value = {
+            **cast(
+                dict[str, SerializedObject],
+                # need to deepcopy to not overwrite the _service_representation dict
+                # when adding a prefix with add_prefix_to_full_access_path
+                deepcopy(self._service_representation["value"]),
+            ),
+            **device_connection_value,
+        }
+
+        return {
+            "full_access_path": "",
+            "name": obj_name,
+            "type": "DeviceConnection",
+            "value": value,
+            "readonly": readonly,
+            "doc": doc,
+        }
+
+    def connect(self) -> None:
+        if not self._sio.reconnection or self._sio.reconnection_attempts > 0:
+            self.reconnect(block_until_connected=False)

src/pydase/client/proxy_loader.py

@@ -351,7 +351,7 @@ class ProxyLoader:
     ) -> Any:
         # Custom types like Components or DataService classes
         component_class = cast(
-            type, Deserializer.get_component_class(serialized_object["type"])
+            type, Deserializer.get_service_base_class(serialized_object["type"])
         )
         class_bases = (
             ProxyClassMixin,

src/pydase/components/device_connection.py

@@ -1,6 +1,7 @@
 import asyncio
 
 import pydase.data_service
+import pydase.task.decorator
 
 
 class DeviceConnection(pydase.data_service.DataService):
@@ -52,7 +53,6 @@ class DeviceConnection(pydase.data_service.DataService):
     def __init__(self) -> None:
         super().__init__()
         self._connected = False
-        self._autostart_tasks["_handle_connection"] = ()  # type: ignore
         self._reconnection_wait_time = 10.0
 
     def connect(self) -> None:
@@ -70,6 +70,7 @@ class DeviceConnection(pydase.data_service.DataService):
         """
         return self._connected
 
+    @pydase.task.decorator.task(autostart=True)
     async def _handle_connection(self) -> None:
         """Automatically tries reconnecting to the device if it is not connected.
         This method leverages the `connect` method and the `connected` property to

src/pydase/data_service/abstract_data_service.py

@@ -1,15 +1,7 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
-
 from pydase.observer_pattern.observable.observable import Observable
 
-if TYPE_CHECKING:
-    from pydase.data_service.data_service import DataService
-    from pydase.data_service.task_manager import TaskManager
-
 
 class AbstractDataService(Observable):
-    __root__: DataService
-    _task_manager: TaskManager
-    _autostart_tasks: dict[str, tuple[Any]]
+    pass

src/pydase/data_service/data_service.py

@@ -5,12 +5,12 @@ from typing import Any
 
 import pydase.units as u
 from pydase.data_service.abstract_data_service import AbstractDataService
-from pydase.data_service.task_manager import TaskManager
 from pydase.observer_pattern.observable.observable import (
     Observable,
 )
 from pydase.utils.helpers import (
     get_class_and_instance_attributes,
+    is_descriptor,
     is_property_attribute,
 )
 from pydase.utils.serialization.serializer import (
@@ -24,11 +24,6 @@ logger = logging.getLogger(__name__)
 class DataService(AbstractDataService):
     def __init__(self) -> None:
         super().__init__()
-        self._task_manager = TaskManager(self)
-
-        if not hasattr(self, "_autostart_tasks"):
-            self._autostart_tasks = {}
-
         self.__check_instance_classes()
 
     def __setattr__(self, __name: str, __value: Any) -> None:
@@ -74,7 +69,7 @@ class DataService(AbstractDataService):
         if not issubclass(
             value_class,
             (int | float | bool | str | list | dict | Enum | u.Quantity | Observable),
-        ):
+        ) and not is_descriptor(__value):
             logger.warning(
                 "Class '%s' does not inherit from DataService. This may lead to"
                 " unexpected behaviour!",

src/pydase/data_service/data_service_observer.py

@@ -8,7 +8,10 @@ from pydase.observer_pattern.observable.observable_object import ObservableObjec
 from pydase.observer_pattern.observer.property_observer import (
     PropertyObserver,
 )
-from pydase.utils.helpers import get_object_attr_from_path
+from pydase.utils.helpers import (
+    get_object_attr_from_path,
+    normalize_full_access_path_string,
+)
 from pydase.utils.serialization.serializer import (
     SerializationPathError,
     SerializedObject,
@@ -53,7 +56,7 @@ class DataServiceObserver(PropertyObserver):
         cached_value = cached_value_dict.get("value")
         if (
             all(part[0] != "_" for part in full_access_path.split("."))
-            and cached_value != value
+            and cached_value != dump(value)["value"]
         ):
             logger.debug("'%s' changed to '%s'", full_access_path, value)
 
@@ -99,7 +102,8 @@ class DataServiceObserver(PropertyObserver):
         )
 
     def _notify_dependent_property_changes(self, changed_attr_path: str) -> None:
-        changed_props = self.property_deps_dict.get(changed_attr_path, [])
+        normalized_attr_path = normalize_full_access_path_string(changed_attr_path)
+        changed_props = self.property_deps_dict.get(normalized_attr_path, [])
         for prop in changed_props:
             # only notify about changing attribute if it is not currently being
             # "changed" e.g. when calling the getter of a property within another

src/pydase/data_service/task_manager.py

@@ -1,225 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import inspect
-import logging
-from enum import Enum
-from typing import TYPE_CHECKING, Any
-
-from pydase.data_service.abstract_data_service import AbstractDataService
-from pydase.utils.helpers import (
-    function_has_arguments,
-    get_class_and_instance_attributes,
-    is_property_attribute,
-)
-
-if TYPE_CHECKING:
-    from collections.abc import Callable
-
-    from .data_service import DataService
-
-logger = logging.getLogger(__name__)
-
-
-class TaskStatus(Enum):
-    RUNNING = "running"
-
-
-class TaskManager:
-    """
-    The TaskManager class is a utility designed to manage asynchronous tasks. It
-    provides functionality for starting, stopping, and tracking these tasks. The class
-    is primarily used by the DataService class to manage its tasks.
-
-    A task in TaskManager is any asynchronous function. To add a task, you simply need
-    to define an async function within your class that extends TaskManager. For example:
-
-    ```python
-    class MyService(DataService):
-        async def my_task(self):
-            # Your task implementation here
-            pass
-    ```
-
-    With the above definition, TaskManager automatically creates `start_my_task` and
-    `stop_my_task` methods that can be used to control the task.
-
-    TaskManager also supports auto-starting tasks. If there are tasks that should start
-    running as soon as an instance of your class is created, you can define them in
-    `self._autostart_tasks` in your class constructor (__init__ method). Here's how:
-
-    ```python
-    class MyService(DataService):
-        def __init__(self):
-            self._autostart_tasks = {
-                "my_task": (*args)  # Replace with actual arguments
-            }
-            self.wait_time = 1
-            super().__init__()
-
-        async def my_task(self, *args):
-            while True:
-                # Your task implementation here
-                await asyncio.sleep(self.wait_time)
-    ```
-
-    In the above example, `my_task` will start running as soon as
-    `_start_autostart_tasks` is called which is done when the DataService instance is
-    passed to the `pydase.Server` class.
-
-    The responsibilities of the TaskManager class are:
-
-    - Track all running tasks: Keeps track of all the tasks that are currently running.
-    This allows for monitoring of task statuses and for making sure tasks do not
-    overlap.
-    - Provide the ability to start and stop tasks: Automatically creates methods to
-    start and stop each task.
-    - Emit notifications when the status of a task changes: Has a built-in mechanism for
-    emitting notifications when a task starts or stops. This is used to update the user
-    interfaces, but can also be used to write logs, etc.
-    """
-
-    def __init__(self, service: DataService) -> None:
-        self.service = service
-
-        self.tasks: dict[str, asyncio.Task[None]] = {}
-        """A dictionary to keep track of running tasks. The keys are the names of the
-        tasks and the values are TaskDict instances which include the task itself and
-        its kwargs.
-        """
-
-        self._set_start_and_stop_for_async_methods()
-
-    @property
-    def _loop(self) -> asyncio.AbstractEventLoop:
-        return asyncio.get_running_loop()
-
-    def _set_start_and_stop_for_async_methods(self) -> None:
-        for name in dir(self.service):
-            # circumvents calling properties
-            if is_property_attribute(self.service, name):
-                continue
-
-            method = getattr(self.service, name)
-            if inspect.iscoroutinefunction(method):
-                if function_has_arguments(method):
-                    logger.info(
-                        "Async function %a is defined with at least one argument. If "
-                        "you want to use it as a task, remove the argument(s) from the "
-                        "function definition.",
-                        method.__name__,
-                    )
-                    continue
-
-                # create start and stop methods for each coroutine
-                setattr(
-                    self.service, f"start_{name}", self._make_start_task(name, method)
-                )
-                setattr(self.service, f"stop_{name}", self._make_stop_task(name))
-
-    def _initiate_task_startup(self) -> None:
-        if self.service._autostart_tasks is not None:
-            for service_name, args in self.service._autostart_tasks.items():
-                start_method = getattr(self.service, f"start_{service_name}", None)
-                if start_method is not None and callable(start_method):
-                    start_method(*args)
-                else:
-                    logger.warning(
-                        "No start method found for service '%s'", service_name
-                    )
-
-    def start_autostart_tasks(self) -> None:
-        self._initiate_task_startup()
-        attrs = get_class_and_instance_attributes(self.service)
-
-        for attr_value in attrs.values():
-            if isinstance(attr_value, AbstractDataService):
-                attr_value._task_manager.start_autostart_tasks()
-            elif isinstance(attr_value, list):
-                for item in attr_value:
-                    if isinstance(item, AbstractDataService):
-                        item._task_manager.start_autostart_tasks()
-
-    def _make_stop_task(self, name: str) -> Callable[..., Any]:
-        """
-        Factory function to create a 'stop_task' function for a running task.
-
-        The generated function cancels the associated asyncio task using 'name' for
-        identification, ensuring proper cleanup. Avoids closure and late binding issues.
-
-        Args:
-            name (str): The name of the coroutine task, used for its identification.
-        """
-
-        def stop_task() -> None:
-            # cancel the task
-            task = self.tasks.get(name, None)
-            if task is not None:
-                self._loop.call_soon_threadsafe(task.cancel)
-
-        return stop_task
-
-    def _make_start_task(
-        self, name: str, method: Callable[..., Any]
-    ) -> Callable[..., Any]:
-        """
-        Factory function to create a 'start_task' function for a coroutine.
-
-        The generated function starts the coroutine as an asyncio task, handling
-        registration and monitoring.
-        It uses 'name' and 'method' to avoid the closure and late binding issue.
-
-        Args:
-            name (str): The name of the coroutine, used for task management.
-            method (callable): The coroutine to be turned into an asyncio task.
-        """
-
-        def start_task() -> None:
-            def task_done_callback(task: asyncio.Task[None], name: str) -> None:
-                """Handles tasks that have finished.
-
-                Removes a task from the tasks dictionary, calls the defined
-                callbacks, and logs and re-raises exceptions."""
-
-                # removing the finished task from the tasks i
-                self.tasks.pop(name, None)
-
-                # emit the notification that the task was stopped
-                self.service._notify_changed(name, None)
-
-                exception = task.exception()
-                if exception is not None:
-                    # Handle the exception, or you can re-raise it.
-                    logger.error(
-                        "Task '%s' encountered an exception: %s: %s",
-                        name,
-                        type(exception).__name__,
-                        exception,
-                    )
-                    raise exception
-
-            async def task() -> None:
-                try:
-                    await method()
-                except asyncio.CancelledError:
-                    logger.info("Task '%s' was cancelled", name)
-
-            if not self.tasks.get(name):
-                # creating the task and adding the task_done_callback which checks
-                # if an exception has occured during the task execution
-                task_object = self._loop.create_task(task())
-                task_object.add_done_callback(
-                    lambda task: task_done_callback(task, name)
-                )
-
-                # Store the task and its arguments in the '__tasks' dictionary. The
-                # key is the name of the method, and the value is a dictionary
-                # containing the task object and the updated keyword arguments.
-                self.tasks[name] = task_object
-
-                # emit the notification that the task was started
-                self.service._notify_changed(name, TaskStatus.RUNNING)
-            else:
-                logger.error("Task '%s' is already running!", name)
-
-        return start_task

src/pydase/frontend/index.html

@@ -6,7 +6,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <meta name="theme-color" content="#000000" />
     <meta name="description" content="Web site displaying a pydase UI." />
-  <script type="module" crossorigin src="/assets/index-D7tStNHJ.js"></script>
+  <script type="module" crossorigin src="/assets/index-BjsjosWf.js"></script>
   <link rel="stylesheet" crossorigin href="/assets/index-D2aktF3W.css">
 </head>
 

src/pydase/observer_pattern/observable/observable.py

@@ -6,7 +6,7 @@ from pydase.observer_pattern.observable.decorators import (
     has_validate_set_decorator,
 )
 from pydase.observer_pattern.observable.observable_object import ObservableObject
-from pydase.utils.helpers import is_property_attribute
+from pydase.utils.helpers import is_descriptor, is_property_attribute
 
 logger = logging.getLogger(__name__)
 
@@ -22,7 +22,9 @@ class Observable(ObservableObject):
             - {"__annotations__"}
         }
         for name, value in class_attrs.items():
-            if isinstance(value, property) or callable(value):
+            if isinstance(value, property) or callable(value) or is_descriptor(value):
+                # Properties, methods and descriptors have to be stored as class
+                # attributes to work properly. So don't make it an instance attribute.
                 continue
             self.__dict__[name] = self._initialise_new_objects(name, value)
 

src/pydase/observer_pattern/observer/property_observer.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from pydase.observer_pattern.observable.observable import Observable
 from pydase.observer_pattern.observer.observer import Observer
+from pydase.utils.helpers import is_descriptor
 
 logger = logging.getLogger(__name__)
 
@@ -60,18 +61,28 @@ class PropertyObserver(Observer):
     def _process_nested_observables_properties(
         self, obj: Observable, deps: dict[str, Any], prefix: str
     ) -> None:
-        for k, value in vars(obj).items():
+        for k, value in {**vars(type(obj)), **vars(obj)}.items():
+            actual_value = value
             prefix = (
                 f"{prefix}." if prefix != "" and not prefix.endswith(".") else prefix
             )
             parent_path = f"{prefix}{k}"
-            if isinstance(value, Observable):
+
+            # Get value from descriptor
+            if not isinstance(value, property) and is_descriptor(value):
+                actual_value = getattr(obj, k)
+
+            if isinstance(actual_value, Observable):
                 new_prefix = f"{parent_path}."
                 deps.update(
-                    self._get_properties_and_their_dependencies(value, new_prefix)
+                    self._get_properties_and_their_dependencies(
+                        actual_value, new_prefix
+                    )
                 )
             elif isinstance(value, list | dict):
-                self._process_collection_item_properties(value, deps, parent_path)
+                self._process_collection_item_properties(
+                    actual_value, deps, parent_path
+                )
 
     def _process_collection_item_properties(
         self,

src/pydase/server/server.py

@@ -13,6 +13,8 @@ from pydase.config import ServiceConfig
 from pydase.data_service.data_service_observer import DataServiceObserver
 from pydase.data_service.state_manager import StateManager
 from pydase.server.web_server import WebServer
+from pydase.task.autostart import autostart_service_tasks
+from pydase.utils.helpers import current_event_loop_exists
 
 HANDLED_SIGNALS = (
     signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
@@ -156,13 +158,18 @@ class Server:
         self._web_port = web_port
         self._enable_web = enable_web
         self._kwargs = kwargs
-        self._loop: asyncio.AbstractEventLoop
         self._additional_servers = additional_servers
         self.should_exit = False
         self.servers: dict[str, asyncio.Future[Any]] = {}
         self._state_manager = StateManager(self._service, filename)
         self._observer = DataServiceObserver(self._state_manager)
         self._state_manager.load_state()
+        autostart_service_tasks(self._service)
+        if not current_event_loop_exists():
+            self._loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(self._loop)
+        else:
+            self._loop = asyncio.get_event_loop()
 
     def run(self) -> None:
         """
@@ -170,7 +177,7 @@ class Server:
 
         This method should be called to start the server after it's been instantiated.
         """
-        asyncio.run(self.serve())
+        self._loop.run_until_complete(self.serve())
 
     async def serve(self) -> None:
         process_id = os.getpid()
@@ -186,10 +193,8 @@ class Server:
         logger.info("Finished server process [%s]", process_id)
 
     async def startup(self) -> None:
-        self._loop = asyncio.get_running_loop()
         self._loop.set_exception_handler(self.custom_exception_handler)
         self.install_signal_handlers()
-        self._service._task_manager.start_autostart_tasks()
 
         for server in self._additional_servers:
             addin_server = server["server"](

src/pydase/task/autostart.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+import pydase.data_service.data_service
+import pydase.task.task
+from pydase.task.task_status import TaskStatus
+from pydase.utils.helpers import is_property_attribute
+
+
+def autostart_service_tasks(
+    service: pydase.data_service.data_service.DataService,
+) -> None:
+    """Starts the service tasks defined with the `autostart` keyword argument.
+
+    This method goes through the attributes of the passed service and its nested
+    [`DataService`][pydase.DataService] instances and calls the start method on
+    autostart-tasks.
+    """
+
+    for attr in dir(service):
+        if is_property_attribute(service, attr) or attr in {
+            "_observers",
+            "__dict__",
+        }:  # prevent eval of property attrs and recursion
+            continue
+
+        val = getattr(service, attr)
+        if isinstance(val, pydase.task.task.Task):
+            if val.autostart and val.status == TaskStatus.NOT_RUNNING:
+                val.start()
+            else:
+                continue
+        else:
+            autostart_nested_service_tasks(val)
+
+
+def autostart_nested_service_tasks(
+    service: pydase.data_service.data_service.DataService | list[Any] | dict[Any, Any],
+) -> None:
+    if isinstance(service, pydase.DataService):
+        autostart_service_tasks(service)
+    elif isinstance(service, list):
+        for entry in service:
+            autostart_nested_service_tasks(entry)
+    elif isinstance(service, dict):
+        for entry in service.values():
+            autostart_nested_service_tasks(entry)

src/pydase/task/decorator.py

@@ -0,0 +1,145 @@
+import logging
+from collections.abc import Callable, Coroutine
+from typing import Any, Generic, TypeVar, overload
+
+from pydase.data_service.data_service import DataService
+from pydase.task.task import Task
+
+logger = logging.getLogger(__name__)
+
+R = TypeVar("R")
+
+
+class PerInstanceTaskDescriptor(Generic[R]):
+    """
+    A descriptor class that provides a unique [`Task`][pydase.task.task.Task] object
+    for each instance of a [`DataService`][pydase.data_service.data_service.DataService]
+    class.
+
+    The `PerInstanceTaskDescriptor` is used to transform an asynchronous function into a
+    task that is managed independently for each instance of a `DataService` subclass.
+    This allows tasks to be initialized, started, and stopped on a per-instance basis,
+    providing better control over task execution within the service.
+
+    The `PerInstanceTaskDescriptor` is not intended to be used directly. Instead, it is
+    used internally by the `@task` decorator to manage task objects for each instance of
+    the service class.
+    """
+
+    def __init__(
+        self,
+        func: Callable[[Any], Coroutine[None, None, R]]
+        | Callable[[], Coroutine[None, None, R]],
+        autostart: bool = False,
+    ) -> None:
+        self.__func = func
+        self.__autostart = autostart
+        self.__task_instances: dict[object, Task[R]] = {}
+
+    def __set_name__(self, owner: type[DataService], name: str) -> None:
+        """Stores the name of the task within the owning class. This method is called
+        automatically when the descriptor is assigned to a class attribute.
+        """
+
+        self.__task_name = name
+
+    @overload
+    def __get__(
+        self, instance: None, owner: type[DataService]
+    ) -> "PerInstanceTaskDescriptor[R]":
+        """Returns the descriptor itself when accessed through the class."""
+
+    @overload
+    def __get__(self, instance: DataService, owner: type[DataService]) -> Task[R]:
+        """Returns the `Task` object associated with the specific `DataService`
+        instance.
+        If no task exists for the instance, a new `Task` object is created and stored
+        in the `__task_instances` dictionary.
+        """
+
+    def __get__(
+        self, instance: DataService | None, owner: type[DataService]
+    ) -> "Task[R] | PerInstanceTaskDescriptor[R]":
+        if instance is None:
+            return self
+
+        # Create a new Task object for this instance, using the function's name.
+        if instance not in self.__task_instances:
+            self.__task_instances[instance] = instance._initialise_new_objects(
+                self.__task_name,
+                Task(self.__func.__get__(instance, owner), autostart=self.__autostart),
+            )
+
+        return self.__task_instances[instance]
+
+
+def task(
+    *, autostart: bool = False
+) -> Callable[
+    [
+        Callable[[Any], Coroutine[None, None, R]]
+        | Callable[[], Coroutine[None, None, R]]
+    ],
+    PerInstanceTaskDescriptor[R],
+]:
+    """
+    A decorator to define an asynchronous function as a per-instance task within a
+    [`DataService`][pydase.DataService] class.
+
+    This decorator transforms an asynchronous function into a
+    [`Task`][pydase.task.task.Task] object that is unique to each instance of the
+    `DataService` class. The resulting `Task` object provides methods like `start()`
+    and `stop()` to control the execution of the task, and manages the task's lifecycle
+    independently for each instance of the service.
+
+    The decorator is particularly useful for defining tasks that need to run
+    periodically or perform asynchronous operations, such as polling data sources,
+    updating databases, or any recurring job that should be managed within the context
+    of a `DataService`.
+    time.
+
+    Args:
+        autostart:
+            If set to True, the task will automatically start when the service is
+            initialized. Defaults to False.
+
+    Returns:
+        A decorator that wraps an asynchronous function in a
+        [`PerInstanceTaskDescriptor`][pydase.task.decorator.PerInstanceTaskDescriptor]
+        object, which, when accessed, provides an instance-specific
+        [`Task`][pydase.task.task.Task] object.
+
+    Example:
+        ```python
+        import asyncio
+
+        import pydase
+        from pydase.task.decorator import task
+
+
+        class MyService(pydase.DataService):
+            @task(autostart=True)
+            async def my_task(self) -> None:
+                while True:
+                    # Perform some periodic work
+                    await asyncio.sleep(1)
+
+
+        if __name__ == "__main__":
+            service = MyService()
+            pydase.Server(service=service).run()
+        ```
+
+        In this example, `my_task` is defined as a task using the `@task` decorator, and
+        it will start automatically when the service is initialized because
+        `autostart=True` is set. You can manually start or stop the task using
+        `service.my_task.start()` and `service.my_task.stop()`, respectively.
+    """
+
+    def decorator(
+        func: Callable[[Any], Coroutine[None, None, R]]
+        | Callable[[], Coroutine[None, None, R]],
+    ) -> PerInstanceTaskDescriptor[R]:
+        return PerInstanceTaskDescriptor(func, autostart=autostart)
+
+    return decorator

src/pydase/task/task.py

@@ -0,0 +1,148 @@
+import asyncio
+import inspect
+import logging
+from collections.abc import Callable, Coroutine
+from typing import (
+    Generic,
+    TypeVar,
+)
+
+import pydase.data_service.data_service
+from pydase.task.task_status import TaskStatus
+from pydase.utils.helpers import current_event_loop_exists
+
+logger = logging.getLogger(__name__)
+
+R = TypeVar("R")
+
+
+class Task(pydase.data_service.data_service.DataService, Generic[R]):
+    """A class representing a task within the `pydase` framework.
+
+    The `Task` class wraps an asynchronous function and provides methods to manage its
+    lifecycle, such as `start()` and `stop()`. It is typically used to perform periodic
+    or recurring jobs in a [`DataService`][pydase.DataService], like reading
+    sensor data, updating databases, or executing other background tasks.
+
+    When a function is decorated with the [`@task`][pydase.task.decorator.task]
+    decorator, it is replaced by a `Task` instance that controls the execution of the
+    original function.
+
+    Args:
+        func:
+            The asynchronous function that this task wraps. It must be a coroutine
+            without arguments.
+        autostart:
+            If set to True, the task will automatically start when the service is
+            initialized. Defaults to False.
+
+    Example:
+        ```python
+        import asyncio
+
+        import pydase
+        from pydase.task.decorator import task
+
+
+        class MyService(pydase.DataService):
+            @task(autostart=True)
+            async def my_task(self) -> None:
+                while True:
+                    # Perform some periodic work
+                    await asyncio.sleep(1)
+
+
+        if __name__ == "__main__":
+            service = MyService()
+            pydase.Server(service=service).run()
+        ```
+
+        In this example, `my_task` is defined as a task using the `@task` decorator, and
+        it will start automatically when the service is initialized because
+        `autostart=True` is set. You can manually start or stop the task using
+        `service.my_task.start()` and `service.my_task.stop()`, respectively.
+    """
+
+    def __init__(
+        self,
+        func: Callable[[], Coroutine[None, None, R | None]],
+        *,
+        autostart: bool = False,
+    ) -> None:
+        super().__init__()
+        self._autostart = autostart
+        self._func_name = func.__name__
+        self._func = func
+        self._task: asyncio.Task[R | None] | None = None
+        self._status = TaskStatus.NOT_RUNNING
+        self._result: R | None = None
+
+        if not current_event_loop_exists():
+            self._loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(self._loop)
+        else:
+            self._loop = asyncio.get_event_loop()
+
+    @property
+    def autostart(self) -> bool:
+        """Defines if the task should be started automatically when the
+        [`Server`][pydase.Server] starts."""
+        return self._autostart
+
+    @property
+    def status(self) -> TaskStatus:
+        """Returns the current status of the task."""
+        return self._status
+
+    def start(self) -> None:
+        """Starts the asynchronous task if it is not already running."""
+        if self._task:
+            return
+
+        def task_done_callback(task: asyncio.Task[R | None]) -> None:
+            """Handles tasks that have finished.
+
+            Updates the task status, calls the defined callbacks, and logs and re-raises
+            exceptions.
+            """
+
+            self._task = None
+            self._status = TaskStatus.NOT_RUNNING
+
+            exception = task.exception()
+            if exception is not None:
+                # Handle the exception, or you can re-raise it.
+                logger.error(
+                    "Task '%s' encountered an exception: %s: %s",
+                    self._func_name,
+                    type(exception).__name__,
+                    exception,
+                )
+                raise exception
+
+            self._result = task.result()
+
+        async def run_task() -> R | None:
+            if inspect.iscoroutinefunction(self._func):
+                logger.info("Starting task %r", self._func_name)
+                self._status = TaskStatus.RUNNING
+                res: Coroutine[None, None, R | None] = self._func()
+                try:
+                    return await res
+                except asyncio.CancelledError:
+                    logger.info("Task '%s' was cancelled", self._func_name)
+                    return None
+            logger.warning(
+                "Cannot start task %r. Function has not been bound yet", self._func_name
+            )
+            return None
+
+        logger.info("Creating task %r", self._func_name)
+        self._task = self._loop.create_task(run_task())
+        self._task.add_done_callback(task_done_callback)
+
+    def stop(self) -> None:
+        """Stops the running asynchronous task by cancelling it."""
+
+        if self._task:
+            self._task.cancel()

src/pydase/task/task_status.py

@@ -0,0 +1,8 @@
+import enum
+
+
+class TaskStatus(enum.Enum):
+    """Possible statuses of a [`Task`][pydase.task.task.Task]."""
+
+    RUNNING = "running"
+    NOT_RUNNING = "not_running"

src/pydase/utils/helpers.py

@@ -114,8 +114,6 @@ def get_class_and_instance_attributes(obj: object) -> dict[str, Any]:
 
     If an attribute exists at both the instance and class level,the value from the
     instance attribute takes precedence.
-    The __root__ object is removed as this will lead to endless recursion in the for
-    loops.
     """
 
     return dict(chain(type(obj).__dict__.items(), obj.__dict__.items()))
@@ -162,6 +160,12 @@ def get_object_attr_from_path(target_obj: Any, path: str) -> Any:
     return get_object_by_path_parts(target_obj, path_parts)
 
 
+def get_task_class() -> type:
+    from pydase.task.task import Task
+
+    return Task
+
+
 def get_component_classes() -> list[type]:
     """
     Returns references to the component classes in a list.
@@ -196,3 +200,48 @@ def function_has_arguments(func: Callable[..., Any]) -> bool:
 
     # Check if there are any parameters left which would indicate additional arguments.
     return len(parameters) > 0
+
+
+def is_descriptor(obj: object) -> bool:
+    """Check if an object is a descriptor."""
+
+    # Exclude functions, methods, builtins and properties
+    if (
+        inspect.isfunction(obj)
+        or inspect.ismethod(obj)
+        or inspect.isbuiltin(obj)
+        or isinstance(obj, property)
+    ):
+        return False
+
+    # Check if it has any descriptor methods
+    return any(hasattr(obj, method) for method in ("__get__", "__set__", "__delete__"))
+
+
+def current_event_loop_exists() -> bool:
+    """Check if an event loop has been set."""
+    import asyncio
+
+    return asyncio.get_event_loop_policy()._local._loop is not None  # type: ignore
+
+
+def normalize_full_access_path_string(s: str) -> str:
+    """Normalizes a string representing a full access path by converting double quotes
+    to single quotes.
+
+    This function is useful for ensuring consistency in strings that represent access
+    paths containing dictionary keys, by replacing all double quotes (`"`) with single
+    quotes (`'`).
+
+    Args:
+        s (str): The input string to be normalized.
+
+    Returns:
+        A new string with all double quotes replaced by single quotes.
+
+    Example:
+        >>> normalize_full_access_path_string('dictionary["first"].my_task')
+        "dictionary['first'].my_task"
+    """
+
+    return s.replace('"', "'")

src/pydase/utils/logging.py

@@ -33,7 +33,7 @@ LOGGING_CONFIG = {
         "default": {
             "formatter": "default",
             "class": "logging.StreamHandler",
-            "stream": "ext://sys.stderr",
+            "stream": "ext://sys.stdout",
         },
     },
     "loggers": {

src/pydase/utils/serialization/deserializer.py

@@ -6,7 +6,9 @@ from typing import TYPE_CHECKING, Any, NoReturn, cast
 import pydase
 import pydase.components
 import pydase.units as u
-from pydase.utils.helpers import get_component_classes
+from pydase.utils.helpers import (
+    get_component_classes,
+)
 from pydase.utils.serialization.types import (
     SerializedDatetime,
     SerializedException,
@@ -49,9 +51,9 @@ class Deserializer:
             return handler(serialized_object)
 
         # Custom types like Components or DataService classes
-        component_class = cls.get_component_class(serialized_object["type"])
-        if component_class:
-            return cls.deserialize_component_type(serialized_object, component_class)
+        service_base_class = cls.get_service_base_class(serialized_object["type"])
+        if service_base_class:
+            return cls.deserialize_data_service(serialized_object, service_base_class)
 
         return None
 
@@ -110,11 +112,11 @@ class Deserializer:
         raise exception(serialized_object["value"])
 
     @staticmethod
-    def get_component_class(type_name: str | None) -> type | None:
+    def get_service_base_class(type_name: str | None) -> type | None:
         for component_class in get_component_classes():
             if type_name == component_class.__name__:
                 return component_class
-        if type_name == "DataService":
+        if type_name in ("DataService", "Task"):
             import pydase
 
             return pydase.DataService
@@ -137,7 +139,7 @@ class Deserializer:
         return property(get, set)
 
     @classmethod
-    def deserialize_component_type(
+    def deserialize_data_service(
         cls, serialized_object: SerializedObject, base_class: type
     ) -> Any:
         def create_proxy_class(serialized_object: SerializedObject) -> type:

src/pydase/utils/serialization/serializer.py

@@ -9,12 +9,14 @@ from typing import TYPE_CHECKING, Any, Literal, cast
 
 import pydase.units as u
 from pydase.data_service.abstract_data_service import AbstractDataService
-from pydase.data_service.task_manager import TaskStatus
+from pydase.task.task_status import TaskStatus
 from pydase.utils.decorators import render_in_frontend
 from pydase.utils.helpers import (
     get_attribute_doc,
     get_component_classes,
     get_data_service_class_reference,
+    get_task_class,
+    is_property_attribute,
     parse_full_access_path,
     parse_serialized_key,
 )
@@ -40,6 +42,8 @@ from pydase.utils.serialization.types import (
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+    from pydase.client.proxy_class import ProxyClass
+
 logger = logging.getLogger(__name__)
 
 
@@ -72,6 +76,7 @@ class Serializer:
         Returns:
             Dictionary representation of `obj`.
         """
+        from pydase.client.client import ProxyClass
 
         result: SerializedObject
 
@@ -81,6 +86,9 @@ class Serializer:
         elif isinstance(obj, datetime):
             result = cls._serialize_datetime(obj, access_path=access_path)
 
+        elif isinstance(obj, ProxyClass):
+            result = cls._serialize_proxy_class(obj, access_path=access_path)
+
         elif isinstance(obj, AbstractDataService):
             result = cls._serialize_data_service(obj, access_path=access_path)
 
@@ -280,6 +288,10 @@ class Serializer:
         if component_base_cls:
             obj_type = component_base_cls.__name__  # type: ignore
 
+        elif isinstance(obj, get_task_class()):
+            # Check if obj is a pydase task
+            obj_type = "Task"
+
         # Get the set of DataService class attributes
         data_service_attr_set = set(dir(get_data_service_class_reference()))
         # Get the set of the object attributes
@@ -294,29 +306,15 @@ class Serializer:
             if key.startswith("_"):
                 continue  # Skip attributes that start with underscore
 
-            # Skip keys that start with "start_" or "stop_" and end with an async
-            # method name
-            if key.startswith(("start_", "stop_")) and key.split("_", 1)[1] in {
-                name
-                for name, _ in inspect.getmembers(
-                    obj, predicate=inspect.iscoroutinefunction
-                )
-            }:
-                continue
-
             val = getattr(obj, key)
 
             path = f"{access_path}.{key}" if access_path else key
             serialized_object = cls.serialize_object(val, access_path=path)
 
-            # If there's a running task for this method
-            if serialized_object["type"] == "method" and key in obj._task_manager.tasks:
-                serialized_object["value"] = TaskStatus.RUNNING.name
-
             value[key] = serialized_object
 
             # If the DataService attribute is a property
-            if isinstance(getattr(obj.__class__, key, None), property):
+            if is_property_attribute(obj, key):
                 prop: property = getattr(obj.__class__, key)
                 value[key]["readonly"] = prop.fset is None
                 value[key]["doc"] = get_attribute_doc(prop)  # overwrite the doc
@@ -330,6 +328,13 @@ class Serializer:
             "doc": doc,
         }
 
+    @classmethod
+    def _serialize_proxy_class(
+        cls, obj: ProxyClass, access_path: str = ""
+    ) -> SerializedDataService:
+        # Get serialization value from the remote service and adapt the full_access_path
+        return add_prefix_to_full_access_path(obj.serialize(), access_path)
+
 
 def dump(obj: Any) -> SerializedObject:
     """Serialize `obj` to a
@@ -580,6 +585,62 @@ def generate_serialized_data_paths(
     return paths
 
 
+def add_prefix_to_full_access_path(
+    serialized_obj: SerializedObject, prefix: str
+) -> Any:
+    """Recursively adds a specified prefix to all full access paths of the serialized
+    object.
+
+    Args:
+        data:
+            The serialized object to process.
+        prefix:
+            The prefix string to prepend to each full access path.
+
+    Returns:
+        The modified serialized object with the prefix added to all full access paths.
+
+    Example:
+        ```python
+        >>> data = {
+        ...     "full_access_path": "",
+        ...     "value": {
+        ...         "item": {
+        ...             "full_access_path": "some_item_path",
+        ...             "value": 1.0
+        ...         }
+        ...     }
+        ... }
+        ...
+        ... modified_data = add_prefix_to_full_access_path(data, 'prefix')
+        {"full_access_path": "prefix", "value": {"item": {"full_access_path":
+        "prefix.some_item_path", "value": 1.0}}}
+        ```
+    """
+
+    try:
+        if serialized_obj.get("full_access_path", None) is not None:
+            serialized_obj["full_access_path"] = (
+                prefix + "." + serialized_obj["full_access_path"]
+                if serialized_obj["full_access_path"] != ""
+                else prefix
+            )
+
+        if isinstance(serialized_obj["value"], list):
+            for value in serialized_obj["value"]:
+                add_prefix_to_full_access_path(cast(SerializedObject, value), prefix)
+
+        elif isinstance(serialized_obj["value"], dict):
+            for value in cast(
+                dict[str, SerializedObject], serialized_obj["value"]
+            ).values():
+                add_prefix_to_full_access_path(cast(SerializedObject, value), prefix)
+    except (TypeError, KeyError, AttributeError):
+        # passed dictionary is not a serialized object
+        pass
+    return serialized_obj
+
+
 def serialized_dict_is_nested_object(serialized_dict: SerializedObject) -> bool:
     value = serialized_dict["value"]
     # We are excluding Quantity here as the value corresponding to the "value" key is

src/pydase/utils/serialization/types.py

@@ -98,7 +98,9 @@ class SerializedException(SerializedObjectBase):
     type: Literal["Exception"]
 
 
-DataServiceTypes = Literal["DataService", "Image", "NumberSlider", "DeviceConnection"]
+DataServiceTypes = Literal[
+    "DataService", "Image", "NumberSlider", "DeviceConnection", "Task"
+]
 
 
 class SerializedDataService(SerializedObjectBase):

tests/client/test_reconnection.py

@@ -0,0 +1,107 @@
+import threading
+from collections.abc import Callable, Generator
+from typing import Any
+
+import pydase
+import pytest
+import socketio.exceptions
+
+
+@pytest.fixture(scope="function")
+def pydase_restartable_server() -> (
+    Generator[
+        tuple[
+            pydase.Server,
+            threading.Thread,
+            pydase.DataService,
+            Callable[
+                [pydase.Server, threading.Thread, pydase.DataService],
+                tuple[pydase.Server, threading.Thread],
+            ],
+        ],
+        None,
+        Any,
+    ]
+):
+    class MyService(pydase.DataService):
+        def __init__(self) -> None:
+            super().__init__()
+            self._name = "MyService"
+            self._my_property = 12.1
+
+        @property
+        def my_property(self) -> float:
+            return self._my_property
+
+        @my_property.setter
+        def my_property(self, value: float) -> None:
+            self._my_property = value
+
+        @property
+        def name(self) -> str:
+            return self._name
+
+    service_instance = MyService()
+    server = pydase.Server(service_instance, web_port=9999)
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+
+    def restart(
+        server: pydase.Server,
+        thread: threading.Thread,
+        service_instance: pydase.DataService,
+    ) -> tuple[pydase.Server, threading.Thread]:
+        server.handle_exit()
+        thread.join()
+
+        server = pydase.Server(service_instance, web_port=9999)
+        new_thread = threading.Thread(target=server.run, daemon=True)
+        new_thread.start()
+
+        return server, new_thread
+
+    yield server, thread, service_instance, restart
+
+    server.handle_exit()
+    thread.join()
+
+
+def test_reconnection(
+    pydase_restartable_server: tuple[
+        pydase.Server,
+        threading.Thread,
+        pydase.DataService,
+        Callable[
+            [pydase.Server, threading.Thread, pydase.DataService],
+            tuple[pydase.Server, threading.Thread],
+        ],
+    ],
+) -> None:
+    client = pydase.Client(
+        url="ws://localhost:9999",
+        sio_client_kwargs={
+            "reconnection": False,
+        },
+    )
+    client_2 = pydase.Client(
+        url="ws://localhost:9999",
+        sio_client_kwargs={
+            "reconnection_attempts": 1,
+        },
+    )
+
+    server, thread, service_instance, restart = pydase_restartable_server
+    service_instance._name = "New service name"
+
+    server, thread = restart(server, thread, service_instance)
+
+    with pytest.raises(socketio.exceptions.BadNamespaceError):
+        client.proxy.name
+        client_2.proxy.name
+
+    client.proxy.reconnect()
+    client_2.proxy.reconnect()
+
+    # the service proxies successfully reconnect and get the new service name
+    assert client.proxy.name == "New service name"
+    assert client_2.proxy.name == "New service name"

tests/components/test_device_connection.py

@@ -3,6 +3,7 @@ import asyncio
 import pydase
 import pydase.components.device_connection
 import pytest
+from pydase.task.autostart import autostart_service_tasks
 from pytest import LogCaptureFixture
 
 
@@ -19,10 +20,9 @@ async def test_reconnection(caplog: LogCaptureFixture) -> None:
             self._connected = True
 
     service_instance = MyService()
+    autostart_service_tasks(service_instance)
 
     assert service_instance._connected is False
 
-    service_instance._task_manager.start_autostart_tasks()
-
     await asyncio.sleep(0.01)
     assert service_instance._connected is True

tests/data_service/test_data_service.py

@@ -36,8 +36,7 @@ def test_unexpected_type_change_warning(caplog: LogCaptureFixture) -> None:
 
 
 def test_basic_inheritance_warning(caplog: LogCaptureFixture) -> None:
-    class SubService(DataService):
-        ...
+    class SubService(DataService): ...
 
     class SomeEnum(Enum):
         HI = 0
@@ -57,11 +56,9 @@ def test_basic_inheritance_warning(caplog: LogCaptureFixture) -> None:
         def name(self) -> str:
             return self._name
 
-        def some_method(self) -> None:
-            ...
+        def some_method(self) -> None: ...
 
-        async def some_task(self) -> None:
-            ...
+        async def some_task(self) -> None: ...
 
     ServiceClass()
 
@@ -129,17 +126,12 @@ def test_exposing_methods(caplog: LogCaptureFixture) -> None:
                 return "some method"
 
     class ClassWithTask(pydase.DataService):
-        async def some_task(self, sleep_time: int) -> None:
-            pass
+        @frontend
+        def some_method(self) -> str:
+            return "some method"
 
     ClassWithTask()
 
-    assert (
-        "Async function 'some_task' is defined with at least one argument. If you want "
-        "to use it as a task, remove the argument(s) from the function definition."
-        in caplog.text
-    )
-
 
 def test_dynamically_added_attribute(caplog: LogCaptureFixture) -> None:
     class MyService(DataService):

tests/data_service/test_data_service_cache.py

@@ -1,7 +1,6 @@
 import logging
 
 import pydase
-import pytest
 from pydase.data_service.data_service_observer import DataServiceObserver
 from pydase.data_service.state_manager import StateManager
 
@@ -33,35 +32,3 @@ def test_nested_attributes_cache_callback() -> None:
         ]
         == "Ciao"
     )
-
-
-@pytest.mark.asyncio(scope="function")
-async def test_task_status_update() -> None:
-    class ServiceClass(pydase.DataService):
-        name = "World"
-
-        async def my_method(self) -> None:
-            pass
-
-    service_instance = ServiceClass()
-    state_manager = StateManager(service_instance)
-    DataServiceObserver(state_manager)
-
-    assert (
-        state_manager.cache_manager.get_value_dict_from_cache("my_method")["type"]
-        == "method"
-    )
-    assert (
-        state_manager.cache_manager.get_value_dict_from_cache("my_method")["value"]
-        is None
-    )
-
-    service_instance.start_my_method()  # type: ignore
-    assert (
-        state_manager.cache_manager.get_value_dict_from_cache("my_method")["type"]
-        == "method"
-    )
-    assert (
-        state_manager.cache_manager.get_value_dict_from_cache("my_method")["value"]
-        == "RUNNING"
-    )

tests/data_service/test_data_service_observer.py

@@ -5,7 +5,7 @@ import pydase
 import pytest
 from pydase.data_service.data_service_observer import DataServiceObserver
 from pydase.data_service.state_manager import StateManager
-from pydase.utils.serialization.serializer import SerializationError
+from pydase.utils.serialization.serializer import SerializationError, dump
 
 logger = logging.getLogger()
 
@@ -146,3 +146,41 @@ def test_private_attribute_does_not_have_to_be_serializable() -> None:
         service_instance.change_publ_attr()
 
     service_instance.change_priv_attr()
+
+
+def test_normalized_attr_path_in_dependent_property_changes(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    class SubService(pydase.DataService):
+        _prop = 10.0
+
+        @property
+        def prop(self) -> float:
+            return self._prop
+
+    class MyService(pydase.DataService):
+        def __init__(self) -> None:
+            super().__init__()
+            self.service_dict = {"one": SubService()}
+
+    service_instance = MyService()
+    state_manager = StateManager(service=service_instance)
+    observer = DataServiceObserver(state_manager=state_manager)
+
+    assert observer.property_deps_dict["service_dict['one']._prop"] == [
+        "service_dict['one'].prop"
+    ]
+
+    # We can use dict key path encoded with double quotes
+    state_manager.set_service_attribute_value_by_path(
+        'service_dict["one"]._prop', dump(11.0)
+    )
+    assert service_instance.service_dict["one"].prop == 11.0
+    assert "'service_dict[\"one\"].prop' changed to '11.0'" in caplog.text
+
+    # We can use dict key path encoded with single quotes
+    state_manager.set_service_attribute_value_by_path(
+        "service_dict['one']._prop", dump(12.0)
+    )
+    assert service_instance.service_dict["one"].prop == 12.0
+    assert "'service_dict[\"one\"].prop' changed to '12.0'" in caplog.text

tests/data_service/test_task_manager.py

@@ -1,135 +0,0 @@
-import asyncio
-import logging
-
-import pydase
-import pytest
-from pydase.data_service.data_service_observer import DataServiceObserver
-from pydase.data_service.state_manager import StateManager
-from pytest import LogCaptureFixture
-
-logger = logging.getLogger("pydase")
-
-
-@pytest.mark.asyncio(scope="function")
-async def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
-    class MyService(pydase.DataService):
-        def __init__(self) -> None:
-            super().__init__()
-            self._autostart_tasks = {  # type: ignore
-                "my_task": (),  # type: ignore
-                "my_other_task": (),  # type: ignore
-            }
-
-        async def my_task(self) -> None:
-            logger.info("Triggered task.")
-
-        async def my_other_task(self) -> None:
-            logger.info("Triggered other task.")
-
-    # Your test code here
-    service_instance = MyService()
-    state_manager = StateManager(service_instance)
-    DataServiceObserver(state_manager)
-    service_instance._task_manager.start_autostart_tasks()
-
-    assert "'my_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-    assert "'my_other_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-
-
-@pytest.mark.asyncio(scope="function")
-async def test_DataService_subclass_autostart_task_callback(
-    caplog: LogCaptureFixture,
-) -> None:
-    class MySubService(pydase.DataService):
-        def __init__(self) -> None:
-            super().__init__()
-            self._autostart_tasks = {  # type: ignore
-                "my_task": (),
-                "my_other_task": (),
-            }
-
-        async def my_task(self) -> None:
-            logger.info("Triggered task.")
-
-        async def my_other_task(self) -> None:
-            logger.info("Triggered other task.")
-
-    class MyService(pydase.DataService):
-        sub_service = MySubService()
-
-    service_instance = MyService()
-    state_manager = StateManager(service_instance)
-    DataServiceObserver(state_manager)
-    service_instance._task_manager.start_autostart_tasks()
-
-    assert "'sub_service.my_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-    assert "'sub_service.my_other_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-
-
-@pytest.mark.asyncio(scope="function")
-async def test_DataService_subclass_list_autostart_task_callback(
-    caplog: LogCaptureFixture,
-) -> None:
-    class MySubService(pydase.DataService):
-        def __init__(self) -> None:
-            super().__init__()
-            self._autostart_tasks = {  # type: ignore
-                "my_task": (),
-                "my_other_task": (),
-            }
-
-        async def my_task(self) -> None:
-            logger.info("Triggered task.")
-
-        async def my_other_task(self) -> None:
-            logger.info("Triggered other task.")
-
-    class MyService(pydase.DataService):
-        sub_services_list = [MySubService() for i in range(2)]
-
-    service_instance = MyService()
-    state_manager = StateManager(service_instance)
-    DataServiceObserver(state_manager)
-    service_instance._task_manager.start_autostart_tasks()
-
-    assert (
-        "'sub_services_list[0].my_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-    )
-    assert (
-        "'sub_services_list[0].my_other_task' changed to 'TaskStatus.RUNNING'"
-        in caplog.text
-    )
-    assert (
-        "'sub_services_list[1].my_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-    )
-    assert (
-        "'sub_services_list[1].my_other_task' changed to 'TaskStatus.RUNNING'"
-        in caplog.text
-    )
-
-
-@pytest.mark.asyncio(scope="function")
-async def test_start_and_stop_task_methods(caplog: LogCaptureFixture) -> None:
-    class MyService(pydase.DataService):
-        def __init__(self) -> None:
-            super().__init__()
-
-        async def my_task(self) -> None:
-            while True:
-                logger.debug("Logging message")
-                await asyncio.sleep(0.1)
-
-    # Your test code here
-    service_instance = MyService()
-    state_manager = StateManager(service_instance)
-    DataServiceObserver(state_manager)
-    service_instance.start_my_task()
-    await asyncio.sleep(0.01)
-
-    assert "'my_task' changed to 'TaskStatus.RUNNING'" in caplog.text
-    assert "Logging message" in caplog.text
-    caplog.clear()
-
-    service_instance.stop_my_task()
-    await asyncio.sleep(0.01)
-    assert "Task 'my_task' was cancelled" in caplog.text

tests/task/test_task.py

@@ -0,0 +1,291 @@
+import asyncio
+import logging
+
+import pydase
+import pytest
+from pydase.data_service.data_service_observer import DataServiceObserver
+from pydase.data_service.state_manager import StateManager
+from pydase.task.autostart import autostart_service_tasks
+from pydase.task.decorator import task
+from pydase.task.task_status import TaskStatus
+from pytest import LogCaptureFixture
+
+logger = logging.getLogger("pydase")
+
+
+@pytest.mark.asyncio(scope="function")
+async def test_start_and_stop_task(caplog: LogCaptureFixture) -> None:
+    class MyService(pydase.DataService):
+        @task()
+        async def my_task(self) -> None:
+            logger.info("Triggered task.")
+            while True:
+                await asyncio.sleep(1)
+
+    # Your test code here
+    service_instance = MyService()
+    state_manager = StateManager(service_instance)
+    DataServiceObserver(state_manager)
+
+    autostart_service_tasks(service_instance)
+    await asyncio.sleep(0.1)
+    assert service_instance.my_task.status == TaskStatus.NOT_RUNNING
+
+    service_instance.my_task.start()
+    await asyncio.sleep(0.1)
+    assert service_instance.my_task.status == TaskStatus.RUNNING
+
+    assert "'my_task.status' changed to 'TaskStatus.RUNNING'" in caplog.text
+    assert "Triggered task." in caplog.text
+    caplog.clear()
+
+    service_instance.my_task.stop()
+    await asyncio.sleep(0.1)
+    assert service_instance.my_task.status == TaskStatus.NOT_RUNNING
+    assert "Task 'my_task' was cancelled" in caplog.text
+
+
+@pytest.mark.asyncio(scope="function")
+async def test_autostart_task(caplog: LogCaptureFixture) -> None:
+    class MyService(pydase.DataService):
+        @task(autostart=True)
+        async def my_task(self) -> None:
+            logger.info("Triggered task.")
+            while True:
+                await asyncio.sleep(1)
+
+    # Your test code here
+    service_instance = MyService()
+    state_manager = StateManager(service_instance)
+    DataServiceObserver(state_manager)
+
+    autostart_service_tasks(service_instance)
+
+    await asyncio.sleep(0.1)
+    assert service_instance.my_task.status == TaskStatus.RUNNING
+
+    assert "'my_task.status' changed to 'TaskStatus.RUNNING'" in caplog.text
+
+
+@pytest.mark.asyncio(scope="function")
+async def test_nested_list_autostart_task(
+    caplog: LogCaptureFixture,
+) -> None:
+    class MySubService(pydase.DataService):
+        @task(autostart=True)
+        async def my_task(self) -> None:
+            logger.info("Triggered task.")
+            while True:
+                await asyncio.sleep(1)
+
+    class MyService(pydase.DataService):
+        sub_services_list = [MySubService() for i in range(2)]
+
+    service_instance = MyService()
+    state_manager = StateManager(service_instance)
+    DataServiceObserver(state_manager)
+    autostart_service_tasks(service_instance)
+
+    await asyncio.sleep(0.1)
+    assert service_instance.sub_services_list[0].my_task.status == TaskStatus.RUNNING
+    assert service_instance.sub_services_list[1].my_task.status == TaskStatus.RUNNING
+
+    assert (
+        "'sub_services_list[0].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+    assert (
+        "'sub_services_list[1].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+
+
+@pytest.mark.asyncio(scope="function")
+async def test_nested_dict_autostart_task(
+    caplog: LogCaptureFixture,
+) -> None:
+    class MySubService(pydase.DataService):
+        @task(autostart=True)
+        async def my_task(self) -> None:
+            logger.info("Triggered task.")
+            while True:
+                await asyncio.sleep(1)
+
+    class MyService(pydase.DataService):
+        sub_services_dict = {"first": MySubService(), "second": MySubService()}
+
+    service_instance = MyService()
+    state_manager = StateManager(service_instance)
+    DataServiceObserver(state_manager)
+
+    autostart_service_tasks(service_instance)
+
+    await asyncio.sleep(0.1)
+
+    assert (
+        service_instance.sub_services_dict["first"].my_task.status == TaskStatus.RUNNING
+    )
+    assert (
+        service_instance.sub_services_dict["second"].my_task.status
+        == TaskStatus.RUNNING
+    )
+
+    assert (
+        "'sub_services_dict[\"first\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"second\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+
+
+@pytest.mark.asyncio(scope="function")
+async def test_manual_start_with_multiple_service_instances(
+    caplog: LogCaptureFixture,
+) -> None:
+    class MySubService(pydase.DataService):
+        @task()
+        async def my_task(self) -> None:
+            logger.info("Triggered task.")
+            while True:
+                await asyncio.sleep(1)
+
+    class MyService(pydase.DataService):
+        sub_services_list = [MySubService() for i in range(2)]
+        sub_services_dict = {"first": MySubService(), "second": MySubService()}
+
+    service_instance = MyService()
+    state_manager = StateManager(service_instance)
+    DataServiceObserver(state_manager)
+
+    autostart_service_tasks(service_instance)
+
+    await asyncio.sleep(0.1)
+
+    assert (
+        service_instance.sub_services_list[0].my_task.status == TaskStatus.NOT_RUNNING
+    )
+    assert (
+        service_instance.sub_services_list[1].my_task.status == TaskStatus.NOT_RUNNING
+    )
+    assert (
+        service_instance.sub_services_dict["first"].my_task.status
+        == TaskStatus.NOT_RUNNING
+    )
+    assert (
+        service_instance.sub_services_dict["second"].my_task.status
+        == TaskStatus.NOT_RUNNING
+    )
+
+    service_instance.sub_services_list[0].my_task.start()
+    await asyncio.sleep(0.01)
+
+    assert service_instance.sub_services_list[0].my_task.status == TaskStatus.RUNNING
+    assert (
+        "'sub_services_list[0].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+    assert (
+        "'sub_services_list[1].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"first\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"second\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+
+    service_instance.sub_services_list[0].my_task.stop()
+    await asyncio.sleep(0.01)
+
+    assert "Task 'my_task' was cancelled" in caplog.text
+    caplog.clear()
+
+    service_instance.sub_services_list[1].my_task.start()
+    await asyncio.sleep(0.01)
+
+    assert service_instance.sub_services_list[1].my_task.status == TaskStatus.RUNNING
+    assert (
+        "'sub_services_list[0].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_list[1].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"first\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"second\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+
+    service_instance.sub_services_list[1].my_task.stop()
+    await asyncio.sleep(0.01)
+
+    assert "Task 'my_task' was cancelled" in caplog.text
+    caplog.clear()
+
+    service_instance.sub_services_dict["first"].my_task.start()
+    await asyncio.sleep(0.01)
+
+    assert (
+        service_instance.sub_services_dict["first"].my_task.status == TaskStatus.RUNNING
+    )
+    assert (
+        "'sub_services_list[0].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_list[1].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"first\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"second\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+
+    service_instance.sub_services_dict["first"].my_task.stop()
+    await asyncio.sleep(0.01)
+
+    assert "Task 'my_task' was cancelled" in caplog.text
+    caplog.clear()
+
+    service_instance.sub_services_dict["second"].my_task.start()
+    await asyncio.sleep(0.01)
+
+    assert (
+        service_instance.sub_services_dict["second"].my_task.status
+        == TaskStatus.RUNNING
+    )
+    assert (
+        "'sub_services_list[0].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_list[1].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"first\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        not in caplog.text
+    )
+    assert (
+        "'sub_services_dict[\"second\"].my_task.status' changed to 'TaskStatus.RUNNING'"
+        in caplog.text
+    )
+
+    service_instance.sub_services_dict["second"].my_task.stop()
+    await asyncio.sleep(0.01)
+
+    assert "Task 'my_task' was cancelled" in caplog.text

tests/utils/serialization/test_serializer.py

@@ -1,4 +1,3 @@
-import asyncio
 import enum
 from datetime import datetime
 from enum import Enum
@@ -8,11 +7,12 @@ import pydase
 import pydase.units as u
 import pytest
 from pydase.components.coloured_enum import ColouredEnum
-from pydase.data_service.task_manager import TaskStatus
+from pydase.task.task_status import TaskStatus
 from pydase.utils.decorators import frontend
 from pydase.utils.serialization.serializer import (
     SerializationPathError,
     SerializedObject,
+    add_prefix_to_full_access_path,
     dump,
     generate_serialized_data_paths,
     get_container_item_by_key,
@@ -214,11 +214,9 @@ async def test_method_serialization() -> None:
             return "some method"
 
         async def some_task(self) -> None:
-            while True:
-                await asyncio.sleep(10)
+            pass
 
     instance = ClassWithMethod()
-    instance.start_some_task()  # type: ignore
 
     assert dump(instance)["value"] == {
         "some_method": {
@@ -234,7 +232,7 @@ async def test_method_serialization() -> None:
         "some_task": {
             "full_access_path": "some_task",
             "type": "method",
-            "value": TaskStatus.RUNNING.name,
+            "value": None,
             "readonly": True,
             "doc": None,
             "async": True,
@@ -1073,3 +1071,156 @@ def test_get_data_paths_from_serialized_object(obj: Any, expected: list[str]) ->
 )
 def test_generate_serialized_data_paths(obj: Any, expected: list[str]) -> None:
     assert generate_serialized_data_paths(dump(obj=obj)["value"]) == expected
+
+
+@pytest.mark.parametrize(
+    "serialized_obj, prefix, expected",
+    [
+        (
+            {
+                "full_access_path": "new_attr",
+                "value": {
+                    "name": {
+                        "full_access_path": "new_attr.name",
+                        "value": "MyService",
+                    }
+                },
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix.new_attr",
+                "value": {
+                    "name": {
+                        "full_access_path": "prefix.new_attr.name",
+                        "value": "MyService",
+                    }
+                },
+            },
+        ),
+        (
+            {
+                "full_access_path": "new_attr",
+                "value": [
+                    {
+                        "full_access_path": "new_attr[0]",
+                        "value": 1.0,
+                    }
+                ],
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix.new_attr",
+                "value": [
+                    {
+                        "full_access_path": "prefix.new_attr[0]",
+                        "value": 1.0,
+                    }
+                ],
+            },
+        ),
+        (
+            {
+                "full_access_path": "new_attr",
+                "value": {
+                    "key": {
+                        "full_access_path": 'new_attr["key"]',
+                        "value": 1.0,
+                    }
+                },
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix.new_attr",
+                "value": {
+                    "key": {
+                        "full_access_path": 'prefix.new_attr["key"]',
+                        "value": 1.0,
+                    }
+                },
+            },
+        ),
+        (
+            {
+                "full_access_path": "new_attr",
+                "value": {"magnitude": 10, "unit": "meter"},
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix.new_attr",
+                "value": {"magnitude": 10, "unit": "meter"},
+            },
+        ),
+        (
+            {
+                "full_access_path": "quantity_list",
+                "value": [
+                    {
+                        "full_access_path": "quantity_list[0]",
+                        "value": {"magnitude": 1.0, "unit": "A"},
+                    }
+                ],
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix.quantity_list",
+                "value": [
+                    {
+                        "full_access_path": "prefix.quantity_list[0]",
+                        "value": {"magnitude": 1.0, "unit": "A"},
+                    }
+                ],
+            },
+        ),
+        (
+            {
+                "full_access_path": "",
+                "value": {
+                    "dict_attr": {
+                        "type": "dict",
+                        "full_access_path": "dict_attr",
+                        "value": {
+                            "foo": {
+                                "full_access_path": 'dict_attr["foo"]',
+                                "type": "dict",
+                                "value": {
+                                    "some_int": {
+                                        "full_access_path": 'dict_attr["foo"].some_int',
+                                        "type": "int",
+                                        "value": 1,
+                                    },
+                                },
+                            },
+                        },
+                    }
+                },
+            },
+            "prefix",
+            {
+                "full_access_path": "prefix",
+                "value": {
+                    "dict_attr": {
+                        "type": "dict",
+                        "full_access_path": "prefix.dict_attr",
+                        "value": {
+                            "foo": {
+                                "full_access_path": 'prefix.dict_attr["foo"]',
+                                "type": "dict",
+                                "value": {
+                                    "some_int": {
+                                        "full_access_path": 'prefix.dict_attr["foo"].some_int',
+                                        "type": "int",
+                                        "value": 1,
+                                    },
+                                },
+                            },
+                        },
+                    }
+                },
+            },
+        ),
+    ],
+)
+def test_add_prefix_to_full_access_path(
+    serialized_obj: SerializedObject, prefix: str, expected: SerializedObject
+) -> None:
+    assert add_prefix_to_full_access_path(serialized_obj, prefix) == expected