Merge pull request #5892 from opsmill/pog-ruff-UP007-20250228

Ruff fixes for UP007

Author: ogenstad

backend/infrahub/lock.py

@@ -5,7 +5,7 @@
 import uuid
 from asyncio import Lock as LocalLock
 from asyncio import sleep
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING
 
 import redis.asyncio as redis
 from prometheus_client import Histogram
@@ -47,7 +47,7 @@
 class InfrahubMultiLock:
     """Context manager to allow multiple locks to be reserved together"""
 
-    def __init__(self, lock_registry: InfrahubLockRegistry, locks: Optional[list[str]] = None) -> None:
+    def __init__(self, lock_registry: InfrahubLockRegistry, locks: list[str] | None = None) -> None:
         self.registry = lock_registry
         self.locks = locks or []
 
@@ -56,9 +56,9 @@ async def __aenter__(self):
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
     ):
         await self.release()
 
@@ -84,9 +84,9 @@ async def __aenter__(self):
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
     ):
         await self.release()
 
@@ -117,18 +117,18 @@ class InfrahubLock:
     def __init__(
         self,
         name: str,
-        connection: Optional[Union[redis.Redis, InfrahubServices]] = None,
-        local: Optional[bool] = None,
+        connection: redis.Redis | InfrahubServices | None = None,
+        local: bool | None = None,
         in_multi: bool = False,
     ) -> None:
         self.use_local: bool = local
         self.local: LocalLock = None
         self.remote: GlobalLock = None
         self.name: str = name
-        self.connection: Optional[redis.Redis] = connection
+        self.connection: redis.Redis | None = connection
         self.in_multi: bool = in_multi
         self.lock_type: str = "multi" if self.in_multi else "individual"
-        self.acquire_time: Optional[int] = None
+        self.acquire_time: int | None = None
         self.event = asyncio.Event()
 
         if not self.connection or (self.use_local is None and name.startswith("local.")):
@@ -146,9 +146,9 @@ async def __aenter__(self):
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
     ):
         await self.release()
 
@@ -179,7 +179,7 @@ async def locked(self) -> bool:
 
 class InfrahubLockRegistry:
     def __init__(
-        self, token: Optional[str] = None, local_only: bool = False, service: Optional[InfrahubServices] = None
+        self, token: str | None = None, local_only: bool = False, service: InfrahubServices | None = None
     ) -> None:
         if config.SETTINGS.cache.enable and not local_only:
             if config.SETTINGS.cache.driver == config.CacheDriver.Redis:
@@ -201,7 +201,7 @@ def __init__(
         self.locks: dict[str, InfrahubLock] = {}
 
     @classmethod
-    def _generate_name(cls, name: str, namespace: Optional[str] = None, local: Optional[bool] = None) -> str:
+    def _generate_name(cls, name: str, namespace: str | None = None, local: bool | None = None) -> str:
         if namespace is None and local is None:
             return name
 
@@ -221,15 +221,15 @@ def get_existing(
         self,
         name: str,
         namespace: str | None,
-        local: Optional[bool] = None,
+        local: bool | None = None,
     ) -> InfrahubLock | None:
         lock_name = self._generate_name(name=name, namespace=namespace, local=local)
         if lock_name not in self.locks:
             return None
         return self.locks[lock_name]
 
     def get(
-        self, name: str, namespace: Optional[str] = None, local: Optional[bool] = None, in_multi: bool = False
+        self, name: str, namespace: str | None = None, local: bool | None = None, in_multi: bool = False
     ) -> InfrahubLock:
         lock_name = self._generate_name(name=name, namespace=namespace, local=local)
         if lock_name not in self.locks:
@@ -252,7 +252,7 @@ def global_graph_lock(self) -> InfrahubMultiLock:
         return InfrahubMultiLock(lock_registry=self, locks=[LOCAL_SCHEMA_LOCK, GLOBAL_GRAPH_LOCK, GLOBAL_SCHEMA_LOCK])
 
 
-def initialize_lock(local_only: bool = False, service: Optional[InfrahubServices] = None) -> None:
+def initialize_lock(local_only: bool = False, service: InfrahubServices | None = None) -> None:
     global registry
     registry = InfrahubLockRegistry(local_only=local_only, service=service)
 

backend/infrahub/message_bus/__init__.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Optional, Union
+from typing import Any
 
 from pydantic import BaseModel, Field
 
@@ -11,21 +11,19 @@
 
 class Meta(BaseModel):
     request_id: str = ""
-    correlation_id: Optional[str] = Field(default=None)
-    reply_to: Optional[str] = Field(default=None)
-    initiator_id: Optional[str] = Field(
+    correlation_id: str | None = Field(default=None)
+    reply_to: str | None = Field(default=None)
+    initiator_id: str | None = Field(
         default=None, description="The worker identity of the initial sender of this message"
     )
-    retry_count: Optional[int] = Field(
-        default=None, description="Indicates how many times this message has been retried."
-    )
-    headers: Optional[dict[str, Any]] = Field(default=None)
-    validator_execution_id: Optional[str] = Field(
+    retry_count: int | None = Field(default=None, description="Indicates how many times this message has been retried.")
+    headers: dict[str, Any] | None = Field(default=None)
+    validator_execution_id: str | None = Field(
         default=None, description="Validator execution ID related to this message"
     )
-    check_execution_id: Optional[str] = Field(default=None, description="Check execution ID related to this message")
+    check_execution_id: str | None = Field(default=None, description="Check execution ID related to this message")
     priority: int = Field(default=3, description="Message Priority")
-    expiration: Optional[int] = Field(default=None, description="TTL before this message expires in seconds")
+    expiration: int | None = Field(default=None, description="TTL before this message expires in seconds")
 
     @classmethod
     def default(cls) -> Meta:
@@ -89,9 +87,9 @@ class InfrahubResponse(InfrahubMessage):
 
     passed: bool = True
     routing_key: str
-    data: Union[dict, InfrahubResponseData] = Field(default_factory=dict)
+    data: dict | InfrahubResponseData = Field(default_factory=dict)
     errors: list[str] = Field(default_factory=list)
-    initial_message: Optional[dict] = Field(
+    initial_message: dict | None = Field(
         default=None,
         description="Initial message in dict format, the primary goal of this field is to provide additional context when there is an error",
     )

backend/infrahub/message_bus/messages/check_generator_run.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from pydantic import Field
 
 from infrahub.generators.models import ProposedChangeGeneratorDefinition
@@ -10,7 +8,7 @@ class CheckGeneratorRun(InfrahubMessage):
     """A check that runs a generator."""
 
     generator_definition: ProposedChangeGeneratorDefinition = Field(..., description="The Generator definition")
-    generator_instance: Optional[str] = Field(
+    generator_instance: str | None = Field(
         default=None, description="The id of the generator instance if it previously existed"
     )
     commit: str = Field(..., description="The commit to target")

backend/infrahub/message_bus/operations/__init__.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 import ujson
 from prefect import Flow
 
@@ -38,7 +36,7 @@
 
 async def execute_message(
     routing_key: str, message_body: bytes, service: InfrahubServices, skip_flow: bool = False
-) -> Optional[MessageTTL]:
+) -> MessageTTL | None:
     message_data = ujson.loads(message_body)
     message = messages.MESSAGE_MAP[routing_key](**message_data)
     message.set_log_data(routing_key=routing_key)

backend/infrahub/message_bus/operations/requests/generator_definition.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from infrahub_sdk.uuidt import UUIDT
 from prefect import flow
 from prefect.logging import get_run_logger
@@ -128,7 +126,7 @@ async def check(message: messages.RequestGeneratorDefinitionCheck, service: Infr
         await service.message_bus.send(message=event)
 
 
-def _run_generator(instance_id: Optional[str], managed_branch: bool, impacted_instances: list[str]) -> bool:
+def _run_generator(instance_id: str | None, managed_branch: bool, impacted_instances: list[str]) -> bool:
     """Returns a boolean to indicate if a generator instance needs to be executed
     Will return true if:
         * The instance_id wasn't set which could be that it's a new object that doesn't have a previous generator instance

backend/infrahub/services/__init__.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Awaitable, Callable, Optional
+from typing import TYPE_CHECKING, Awaitable, Callable
 
 from infrahub.components import ComponentType
 from infrahub.exceptions import InitializationError
@@ -182,7 +182,7 @@ async def shutdown(self) -> None:
         await self.scheduler.shutdown()
         await self.message_bus.shutdown()
 
-    async def send(self, message: InfrahubMessage, delay: Optional[MessageTTL] = None, is_retry: bool = False) -> None:
+    async def send(self, message: InfrahubMessage, delay: MessageTTL | None = None, is_retry: bool = False) -> None:
         routing_key = ROUTING_KEY_MAP.get(type(message))
         if not routing_key:
             raise ValueError("Unable to determine routing key")

backend/infrahub/services/adapters/cache/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from infrahub.message_bus.types import KVTTL
@@ -16,12 +16,12 @@ async def delete(self, key: str) -> None:
         raise NotImplementedError()
 
     @abstractmethod
-    async def get(self, key: str) -> Optional[str]:
+    async def get(self, key: str) -> str | None:
         """Retrieve a value from the cache."""
         raise NotImplementedError()
 
     @abstractmethod
-    async def get_values(self, keys: list[str]) -> list[Optional[str]]:
+    async def get_values(self, keys: list[str]) -> list[str | None]:
         """Return a list the values for requested keys."""
         raise NotImplementedError()
 
@@ -31,8 +31,6 @@ async def list_keys(self, filter_pattern: str) -> list[str]:
         raise NotImplementedError()
 
     @abstractmethod
-    async def set(
-        self, key: str, value: str, expires: Optional[KVTTL] = None, not_exists: bool = False
-    ) -> Optional[bool]:
+    async def set(self, key: str, value: str, expires: KVTTL | None = None, not_exists: bool = False) -> bool | None:
         """Set a value in the cache."""
         raise NotImplementedError()

backend/infrahub/services/adapters/cache/nats.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import ssl
-from typing import Optional
 
 import nats
 
@@ -84,7 +83,7 @@ async def delete(self, key: str) -> None:
         key = self._tokenize_key_name(key)
         await self._get_kv(key).delete(key)
 
-    async def get(self, key: str) -> Optional[str]:
+    async def get(self, key: str) -> str | None:
         key = self._tokenize_key_name(key)
         try:
             entry = await self._get_kv(key).get(key=key)
@@ -94,7 +93,7 @@ async def get(self, key: str) -> Optional[str]:
             pass
         return None
 
-    async def get_values(self, keys: list[str]) -> list[Optional[str]]:
+    async def get_values(self, keys: list[str]) -> list[str | None]:
         return [await self.get(key) for key in keys]
 
     async def _keys(self, kv: nats.js.kv.KeyValue, filter_pattern: str) -> list[str]:
@@ -138,9 +137,9 @@ async def set(
         self,
         key: str,
         value: str,
-        expires: Optional[KVTTL] = None,  # noqa: ARG002
+        expires: KVTTL | None = None,  # noqa: ARG002
         not_exists: bool = False,
-    ) -> Optional[bool]:
+    ) -> bool | None:
         key = self._tokenize_key_name(key)
         if not_exists:
             try:

backend/infrahub/services/adapters/cache/redis.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 import redis.asyncio as redis
 
 from infrahub import config
@@ -22,13 +20,13 @@ def __init__(self) -> None:
     async def delete(self, key: str) -> None:
         await self.connection.delete(key)
 
-    async def get(self, key: str) -> Optional[str]:
+    async def get(self, key: str) -> str | None:
         value = await self.connection.get(name=key)
         if value is not None:
             return value.decode()
         return None
 
-    async def get_values(self, keys: list[str]) -> list[Optional[str]]:
+    async def get_values(self, keys: list[str]) -> list[str | None]:
         values = await self.connection.mget(keys=keys)
         return [value.decode() if value is not None else value for value in values]
 
@@ -44,7 +42,5 @@ async def list_keys(self, filter_pattern: str) -> list[str]:
 
         return [key.decode() for key in keys]
 
-    async def set(
-        self, key: str, value: str, expires: Optional[KVTTL] = None, not_exists: bool = False
-    ) -> Optional[bool]:
+    async def set(self, key: str, value: str, expires: KVTTL | None = None, not_exists: bool = False) -> bool | None:
         return await self.connection.set(name=key, value=value, ex=expires.value if expires else None, nx=not_exists)

backend/infrahub/services/adapters/message_bus/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional, TypeVar
+from typing import TYPE_CHECKING, TypeVar
 
 from infrahub.message_bus.messages import ROUTING_KEY_MAP
 
@@ -36,7 +36,7 @@ async def shutdown(self) -> None:  # noqa: B027 We want a default empty behavior
 
     @abstractmethod
     async def publish(
-        self, message: InfrahubMessage, routing_key: str, delay: Optional[MessageTTL] = None, is_retry: bool = False
+        self, message: InfrahubMessage, routing_key: str, delay: MessageTTL | None = None, is_retry: bool = False
     ) -> None:
         raise NotImplementedError()
 
@@ -48,7 +48,7 @@ async def reply(self, message: InfrahubMessage, routing_key: str) -> None:
     async def rpc(self, message: InfrahubMessage, response_class: type[ResponseClass]) -> ResponseClass:
         raise NotImplementedError()
 
-    async def send(self, message: InfrahubMessage, delay: Optional[MessageTTL] = None, is_retry: bool = False) -> None:
+    async def send(self, message: InfrahubMessage, delay: MessageTTL | None = None, is_retry: bool = False) -> None:
         routing_key = ROUTING_KEY_MAP.get(type(message))
         if not routing_key:
             raise ValueError("Unable to determine routing key")