Enable pyupgrade in ruff

asyncua/common/callback.py

@@ -25,7 +25,7 @@ class CallbackType(Enum):
  PostRead = 7
 
 
-class Callback(object):
+class Callback:
  def __init__(self):
  self.__name = None
 
@@ -44,12 +44,12 @@ def __init__(self, request_params, response_params, user=None, is_external=False
  self.user = user
 
 
-class CallbackSubscriberInterface(object):
+class CallbackSubscriberInterface:
  def getSubscribedEvents(self):
  raise NotImplementedError()
 
 
-class CallbackService(object):
+class CallbackService:
  def __init__(self):
  self._listeners = {}
 

asyncua/common/statemachine.py

@@ -72,7 +72,7 @@ def __init__(self, id, name: str = None, number: int = None, node: Node = None):
  self.node: Node = node # will be written from statemachine.add_state() or you need to overwrite it if the state is part of xml
 
 
-class StateMachine(object):
+class StateMachine:
  '''
  Implementation of an StateMachineType (most basic type)
  CurrentState: Mandatory "StateVariableType"

asyncua/common/structures.py

@@ -21,7 +21,7 @@
 _logger = logging.getLogger(__name__)
 
 
-class EnumType(object):
+class EnumType:
  def __init__(self, name):
  self.name = clean_name(name)
  self.fields = []
@@ -109,7 +109,7 @@ class {self.name}:
  return code
 
 
-class Field(object):
+class Field:
  def __init__(self, name):
  self.name = name
  self.uatype = None
@@ -123,7 +123,7 @@ def __str__(self):
  __repr__ = __str__
 
 
-class StructGenerator(object):
+class StructGenerator:
  def __init__(self):
  self.model = []
 

asyncua/crypto/security_policies.py

@@ -25,7 +25,7 @@ def require_cryptography(obj):
  raise UaError(f"Can't use {obj.__class__.__name__}, cryptography module is not installed")
 
 
-class Signer(object):
+class Signer:
  """
  Abstract base class for cryptographic signature algorithm
  """
@@ -41,7 +41,7 @@ def signature(self, data):
  pass
 
 
-class Verifier(object):
+class Verifier:
  """
  Abstract base class for cryptographic signature verification
  """
@@ -62,7 +62,7 @@ def reset(self):
  attrs[k] = None
 
 
-class Encryptor(object):
+class Encryptor:
  """
  Abstract base class for encryption algorithm
  """
@@ -82,7 +82,7 @@ def encrypt(self, data):
  pass
 
 
-class Decryptor(object):
+class Decryptor:
  """
  Abstract base class for decryption algorithm
  """

asyncua/server/address_space.py

@@ -34,14 +34,14 @@
 _logger = logging.getLogger(__name__)
 
 
-class AttributeValue(object):
+class AttributeValue:
  """
  The class holds the value(s) of an attribute and callbacks.
  """
  def __init__(self, value: ua.DataValue):
  self.value: Optional[ua.DataValue] = value
  self.value_callback: Optional[Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue]] = None
- self.value_setter: Optional[Callable[["NodeData", ua.AttributeIds, ua.DataValue], None]] = None
+ self.value_setter: Optional[Callable[[NodeData, ua.AttributeIds, ua.DataValue], None]] = None
  self.datachange_callbacks = {}
 
  def __str__(self) -> str:
@@ -112,7 +112,7 @@ async def write(self, params: ua.WriteParameters, user: User = User(role=UserRol
  return res
 
 
-class ViewService(object):
+class ViewService:
  """
  This class implements the view service set defined in the opc ua standard.
  https://reference.opcfoundation.org/v104/Core/docs/Part4/5.8.1/

asyncua/server/history_sql.py

@@ -228,7 +228,7 @@ async def _get_event_fields(self, evtypes):
  # get all fields from the event types that are to be historized
  ev_aggregate_fields = []
  for event_type in evtypes:
- ev_aggregate_fields.extend((await get_event_properties_from_type_node(event_type)))
+ ev_aggregate_fields.extend(await get_event_properties_from_type_node(event_type))
  ev_fields = []
  for field in set(ev_aggregate_fields):
  ev_fields.append((await field.read_display_name()).Text)

asyncua/server/internal_session.py

@@ -30,10 +30,10 @@ class InternalSession(AbstractSession):
  _counter = 10
  _auth_counter = 1000
 
- def __init__(self, internal_server: "InternalServer", aspace: AddressSpace, submgr: SubscriptionService, name,
+ def __init__(self, internal_server: InternalServer, aspace: AddressSpace, submgr: SubscriptionService, name,
  user=User(role=UserRole.Anonymous), external=False):
  self.logger = logging.getLogger(__name__)
- self.iserver: "InternalServer" = internal_server
+ self.iserver: InternalServer = internal_server
  # define if session is external, we need to copy some objects if it is internal
  self.external = external
  self.aspace: AddressSpace = aspace

asyncua/sync.py

@@ -155,7 +155,7 @@ def sync_uaclient_method(aio_func):
  ...
  ```
  """
- def sync_method(client: 'Client'):
+ def sync_method(client: Client):
  uaclient = client.aio_obj.uaclient
  return functools.partial(sync_wrapper(aio_func), client.tloop, uaclient)
 
@@ -176,7 +176,7 @@ def sync_async_client_method(aio_func):
  ...
  ```
  """
- def sync_method(client: 'Client'):
+ def sync_method(client: Client):
  return functools.partial(sync_wrapper(aio_func), client.tloop, client)
 
  return sync_method
@@ -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.SubscriptionHandler, 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)
@@ -382,15 +382,15 @@ def get_endpoints(self) -> List[ua.EndpointDescription]: # type: ignore[empty-b
  @syncmethod
  def register_server(
  self,
- server: "Server",
+ server: Server,
  discovery_configuration: Optional[ua.DiscoveryConfiguration] = None,
  ) -> None:
  pass
 
  @syncmethod
  def unregister_server(
  self,
- server: "Server",
+ server: Server,
  discovery_configuration: Optional[ua.DiscoveryConfiguration] = None,
  ) -> None:
  pass
@@ -969,7 +969,7 @@ def get_path(self, max_length: int = 20, as_string: Literal[False] = False) -> L
  ...
 
  @overload
- def get_path(self, max_length: int = 20, as_string: Literal[True] = True) -> List["str"]:
+ def get_path(self, max_length: int = 20, as_string: Literal[True] = True) -> List[str]:
  ...
 
  @syncmethod
@@ -1157,7 +1157,7 @@ def new_struct_field(
 
 @syncfunc(aio_func=common.structures104.new_enum)
 def new_enum( # type: ignore[empty-body]
- server: Union["Server", "Client"],
+ server: Union[Server, Client],
  idx: Union[int, ua.NodeId],
  name: Union[int, ua.QualifiedName],
  values: List[str],
@@ -1168,7 +1168,7 @@ def new_enum( # type: ignore[empty-body]
 
 @syncfunc(aio_func=common.structures104.new_struct)
 def new_struct( # type: ignore[empty-body]
- server: Union["Server", "Client"],
+ server: Union[Server, Client],
  idx: Union[int, ua.NodeId],
  name: Union[int, ua.QualifiedName],
  fields: List[ua.StructureField],

asyncua/tools.py

@@ -462,7 +462,7 @@ async def _lsprint_long(pnode, depth, indent=""):
  await _lsprint_long(node, depth - 1, indent + " ")
 
 
-class SubHandler(object):
+class SubHandler:
  def datachange_notification(self, node, val, data):
  print("New data change event", node, val, data)
 

examples/client-example.py

@@ -6,7 +6,7 @@
 _logger = logging.getLogger(__name__)
 
 
-class SubHandler(object):
+class SubHandler:
  """
  Subscription Handler. To receive events from server for a subscription
  data_change and event methods are called directly from receiving thread.

examples/client_to_kepware.py

@@ -6,7 +6,7 @@
 from asyncua import Client
 
 
-class SubHandler(object):
+class SubHandler:
 
  """
  Subscription Handler. To receive events from server for a subscription

examples/client_to_prosys.py

@@ -4,7 +4,7 @@
 from asyncua import Client
 
 
-class SubHandler(object):
+class SubHandler:
  """
  Subscription Handler. To receive events from server for a subscription
  """

examples/server-example.py

@@ -12,7 +12,7 @@
 _logger = logging.getLogger(__name__)
 
 
-class SubHandler(object):
+class SubHandler:
 
  """
  Subscription Handler. To receive events from server for a subscription

examples/server-ua-python-mirror.py

@@ -17,7 +17,7 @@
 # Be aware that subscription calls are asynchronous.
 
 
-class SubHandler(object):
+class SubHandler:
  """
  Subscription Handler. To receive events from server for a subscription.
  The handler forwards updates to it's referenced python object
@@ -33,7 +33,7 @@ def datachange_notification(self, node, val, data):
  setattr(self.obj, _node_name.Name, data.monitored_item.Value.Value.Value)
 
 
-class UaObject(object):
+class UaObject:
  """
  Python object which mirrors an OPC UA object
  Child UA variables/properties are auto subscribed to to synchronize python with UA server

examples/sync/client-example.py

@@ -18,7 +18,7 @@ def embed():
 from asyncua.sync import Client, ThreadLoop
 
 
-class SubHandler(object):
+class SubHandler:
 
  """
  Subscription Handler. To receive events from server for a subscription

examples/sync/client_to_prosys.py

@@ -3,7 +3,7 @@
 from asyncua.sync import Client
 
 
-class SubHandler(object):
+class SubHandler:
 
  """
  Client to subscription. It will receive events from server

examples/sync/server-example.py

@@ -23,7 +23,7 @@ def embed():
 from asyncua.sync import Server, ThreadLoop
 
 
-class SubHandler(object):
+class SubHandler:
 
  """
  Subscription Handler. To receive events from server for a subscription

ruff.toml

@@ -1,6 +1,17 @@
-lint.select = ["E", "F", "G004", "W"]
 extend-exclude = ["docs", "nodeset", "schemas", "*_auto.py", "event_objects.py", "standard_address_space_services.py"]
 line-length = 999
+target-version = "py37"
+[lint]
+select = ["E", "F", "G004", "W", "UP"]
+ignore = [
+ "UP032", # Use f-string instead of `format` call
+ "UP030", # Use implicit references for positional format fields
+ "UP027", # Replace unpacked list comprehension with a generator expression
+ # The following can be removed once the minimum supported Python version is 3.10
+ # https:/astral-sh/ruff/issues/5035
+ "UP006", # Use `list` instead of `List` for type annotation
+ "UP007", # Use `X | Y` for type annotations
+]
 [lint.per-file-ignores]
 "__init__.py" = ["F401", "F403"]
 "examples/*" = ["F841", "E402"]

test_external_server.py

@@ -28,7 +28,7 @@ def event_notification(self, event):
  self.future.set_result(event)
 
 
-class MySubHandler2(object):
+class MySubHandler2:
  def __init__(self):
  self.results = []
 

tests/test_common.py

@@ -1,4 +1,3 @@
-# encoding: utf-8
 
 """
 Tests that will be run twice. Once on server side and once on
@@ -740,7 +739,7 @@ async def test_value(opc):
  assert 1.98 == await v.read_value()
  dvar = ua.DataValue(var)
  dv = await v.read_data_value()
- assert ua.DataValue == type(dv)
+ assert ua.DataValue is type(dv)
  assert dvar.Value == dv.Value
  assert dvar.Value == var
  await opc.opc.delete_nodes([v])

tests/test_connections.py

@@ -1,4 +1,3 @@
-# coding: utf-8
 import asyncio
 import pytest
 import struct

tests/test_subscriptions.py

@@ -1063,7 +1063,7 @@ def set_result(self, publish_result):
  if publish_result.NotificationMessage.NotificationData is not None:
  for notif in publish_result.NotificationMessage.NotificationData:
  if isinstance(notif, ua.DataChangeNotification):
- values.extend((item.Value.Value.Value for item in notif.MonitoredItems))
+ values.extend(item.Value.Value.Value for item in notif.MonitoredItems)
  self.fut.set_result(values)
 
  async def result(self):

tests/test_unit.py

@@ -1,4 +1,3 @@
-# encoding: utf-8
 # ! /usr/bin/env python
 """
 Simple unit test that do not need to setup a server or a client
@@ -519,7 +518,7 @@ def test_null_string():
 def test_empty_extension_object():
  obj = ua.ExtensionObject()
  obj2 = extensionobject_from_binary(ua.utils.Buffer(extensionobject_to_binary(obj)))
- assert type(obj) == type(obj2)
+ assert type(obj) is type(obj2)
  assert obj == obj2
 
 
@@ -528,12 +527,12 @@ def test_extension_object():
  obj.UserName = "admin"
  obj.Password = b"pass"
  obj2 = extensionobject_from_binary(ua.utils.Buffer(extensionobject_to_binary(obj)))
- assert type(obj) == type(obj2)
+ assert type(obj) is type(obj2)
  assert obj.UserName == obj2.UserName
  assert obj.Password == obj2.Password
  v1 = ua.Variant(obj)
  v2 = variant_from_binary(ua.utils.Buffer(variant_to_binary(v1)))
- assert type(v1) == type(v2)
+ assert type(v1) is type(v2)
  assert v1.VariantType == v2.VariantType
 
 
@@ -545,7 +544,7 @@ def test_unknown_extension_object():
 
  data = ua.utils.Buffer(extensionobject_to_binary(obj))
  obj2 = extensionobject_from_binary(data)
- assert type(obj2) == ua.ExtensionObject
+ assert type(obj2) is ua.ExtensionObject
  assert obj2.TypeId == obj.TypeId
  assert obj2.Body == b'example of data in custom format'