312

.github/workflows/python-package.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.7', '3.11', 'pypy-3.8']
+        python-version: ['3.7', '3.12', 'pypy-3.10']
 
     steps:
     - uses: actions/checkout@v4

asyncua/common/subscription.py

@@ -81,35 +81,41 @@ def status_change_notification(self, status: ua.StatusChangeNotification) -> Non
         ...
 
 
-SubscriptionHandler = Union[DataChangeNotificationHandler, EventNotificationHandler, StatusChangeNotificationHandler]
-"""
-Protocol class representing subscription handlers to receive events from server.
-"""
-
-
-class SubHandler:
-    """
-    Subscription Handler. To receive events from server for a subscription
-    This class is just a sample class. Whatever class having these methods can be used
-    """
-
-    def datachange_notification(self, node: Node, val: Any, data: DataChangeNotif) -> None:
+class DataChangeNotificationHandlerAsync(Protocol):
+    async def datachange_notification(self, node: Node, val: Any, data: DataChangeNotif) -> None:
         """
         called for every datachange notification from server
         """
-        pass
+        ...
 
-    def event_notification(self, event: ua.EventNotificationList) -> None:
+
+class EventNotificationHandlerAsync(Protocol):
+    async def event_notification(self, event: ua.EventNotificationList) -> None:
         """
         called for every event notification from server
         """
-        pass
+        ...
 
-    def status_change_notification(self, status: ua.StatusChangeNotification) -> None:
+
+class StatusChangeNotificationHandlerAsync(Protocol):
+    async def status_change_notification(self, status: ua.StatusChangeNotification) -> None:
         """
         called for every status change notification from server
         """
-        pass
+        ...
+
+
+SubscriptionHandler = Union[
+        DataChangeNotificationHandler,
+        EventNotificationHandler,
+        StatusChangeNotificationHandler,
+        DataChangeNotificationHandlerAsync,
+        EventNotificationHandlerAsync,
+        StatusChangeNotificationHandlerAsync,
+        ]
+"""
+Protocol class representing subscription handlers to receive events from server.
+"""
 
 
 class Subscription:

asyncua/server/history.py

@@ -7,7 +7,6 @@
 from datetime import timezone
 
 from asyncua import ua
-from asyncua.common import subscription
 from asyncua.common.subscription import Subscription, SubscriptionHandler
 from ..common.utils import Buffer
 
@@ -217,7 +216,7 @@ async def stop(self):
         pass
 
 
-class SubHandler(subscription.SubHandler):
+class SubHandler:
     def __init__(self, storage: HistoryStorageInterface):
         self.storage = storage
 

asyncua/sync.py

@@ -322,7 +322,7 @@ def set_security_string(self, string: str) -> None:
     def load_enums(self) -> Dict[str, Type]:  # type: ignore[empty-body]
         pass
 
-    def create_subscription(self, period: Union[ua.CreateSubscriptionParameters, float], handler: subscription.SubHandler, publishing: bool = True) -> "Subscription":
+    def create_subscription(self, period: Union[ua.CreateSubscriptionParameters, float], handler: subscription.SubscriptionHandler, publishing: bool = True) -> "Subscription":
         coro = self.aio_obj.create_subscription(period, _SubHandler(self.tloop, handler), publishing)
         aio_sub = self.tloop.post(coro)
         return Subscription(self.tloop, aio_sub)