Additional fixes for ruff UP007

Author: ogenstad

backend/infrahub/api/dependencies.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, AsyncIterator, Optional
+from typing import TYPE_CHECKING, AsyncIterator
 
 from fastapi import Depends, Query, Request
 from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer
@@ -25,7 +25,7 @@
 api_key_scheme = APIKeyHeader(name="X-INFRAHUB-KEY", auto_error=False)
 
 
-async def cookie_auth_scheme(request: Request) -> Optional[str]:
+async def cookie_auth_scheme(request: Request) -> str | None:
     return request.cookies.get("access_token")  # Replace with the actual name of your JWT cookie
 
 
@@ -62,7 +62,7 @@ async def get_access_token(
 async def get_refresh_token(
     request: Request,
     db: InfrahubDatabase = Depends(get_db),
-    jwt_header: Optional[HTTPAuthorizationCredentials] = Depends(jwt_scheme),
+    jwt_header: HTTPAuthorizationCredentials | None = Depends(jwt_scheme),
 ) -> RefreshTokenData:
     token = None
 
@@ -83,8 +83,8 @@ async def get_refresh_token(
 
 async def get_branch_params(
     db: InfrahubDatabase = Depends(get_db),
-    branch_name: Optional[str] = Query(None, alias="branch", description="Name of the branch to use for the query"),
-    at: Optional[str] = Query(None, description="Time to use for the query, in absolute or relative format"),
+    branch_name: str | None = Query(None, alias="branch", description="Name of the branch to use for the query"),
+    at: str | None = Query(None, description="Time to use for the query, in absolute or relative format"),
 ) -> BranchParams:
     branch = await registry.get_branch(db=db, branch=branch_name)
 
@@ -93,7 +93,7 @@ async def get_branch_params(
 
 async def get_branch_dep(
     db: InfrahubDatabase = Depends(get_db),
-    branch_name: Optional[str] = Query(None, alias="branch", description="Name of the branch to use for the query"),
+    branch_name: str | None = Query(None, alias="branch", description="Name of the branch to use for the query"),
 ) -> Branch:
     return await registry.get_branch(db=db, branch=branch_name)
 

backend/infrahub/api/diff/validation_models.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any
 
 from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 
@@ -9,13 +9,13 @@
 class DiffQueryValidated(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
     branch: Branch
-    time_from: Optional[str] = None
-    time_to: Optional[str] = None
+    time_from: str | None = None
+    time_to: str | None = None
     branch_only: bool
 
     @field_validator("time_from", "time_to", mode="before")
     @classmethod
-    def validate_time(cls, value: Optional[str]) -> Optional[str]:
+    def validate_time(cls, value: str | None) -> str | None:
         if not value:
             return None
         Timestamp(value)
@@ -24,12 +24,12 @@ def validate_time(cls, value: Optional[str]) -> Optional[str]:
     @model_validator(mode="before")
     @classmethod
     def validate_time_from_if_required(cls, values: dict[str, Any]) -> dict[str, Any]:
-        branch: Optional[Branch] = values.get("branch")
-        time_from: Optional[Timestamp] = values.get("time_from")
+        branch: Branch | None = values.get("branch")
+        time_from: Timestamp | None = values.get("time_from")
         if getattr(branch, "is_default", False) and not time_from:
             branch_name = getattr(branch, "name", "")
             raise ValueError(f"time_from is mandatory when diffing on the default branch `{branch_name}`.")
-        time_to: Optional[Timestamp] = values.get("time_to")
+        time_to: Timestamp | None = values.get("time_to")
         if time_to and time_from and time_to < time_from:
             raise ValueError("time_from and time_to are not a valid time range")
         return values

backend/infrahub/database/__init__.py

@@ -3,7 +3,7 @@
 import asyncio
 import random
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Callable, Coroutine, Optional, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Callable, Coroutine, TypeVar
 
 from neo4j import (
     READ_ACCESS,
@@ -69,7 +69,7 @@ class InfrahubDatabaseSessionMode(InfrahubStringEnum):
     WRITE = "write"
 
 
-def get_branch_name(branch: Optional[Union[Branch, str]] = None) -> str:
+def get_branch_name(branch: Branch | str | None = None) -> str:
     if not branch:
         return registry.default_branch
     if isinstance(branch, str):
@@ -82,43 +82,39 @@ class DatabaseSchemaManager:
     def __init__(self, db: InfrahubDatabase) -> None:
         self._db = db
 
-    def get(self, name: str, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True) -> MainSchemaTypes:
+    def get(self, name: str, branch: Branch | str | None = None, duplicate: bool = True) -> MainSchemaTypes:
         branch_name = get_branch_name(branch=branch)
         if branch_name not in self._db._schemas:
             return registry.schema.get(name=name, branch=branch, duplicate=duplicate)
         return self._db._schemas[branch_name].get(name=name, duplicate=duplicate)
 
-    def get_node_schema(
-        self, name: str, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True
-    ) -> NodeSchema:
+    def get_node_schema(self, name: str, branch: Branch | str | None = None, duplicate: bool = True) -> NodeSchema:
         schema = self.get(name=name, branch=branch, duplicate=duplicate)
         if schema.is_node_schema:
             return schema
 
         raise ValueError("The selected node is not of type NodeSchema")
 
-    def set(self, name: str, schema: MainSchemaTypes, branch: Optional[str] = None) -> int:
+    def set(self, name: str, schema: MainSchemaTypes, branch: str | None = None) -> int:
         branch_name = get_branch_name(branch=branch)
         if branch_name not in self._db._schemas:
             return registry.schema.set(name=name, schema=schema, branch=branch)
         return self._db._schemas[branch_name].set(name=name, schema=schema)
 
-    def has(self, name: str, branch: Optional[Union[Branch, str]] = None) -> bool:
+    def has(self, name: str, branch: Branch | str | None = None) -> bool:
         branch_name = get_branch_name(branch=branch)
         if branch_name not in self._db._schemas:
             return registry.schema.has(name=name, branch=branch)
         return self._db._schemas[branch_name].has(name=name)
 
-    def get_full(
-        self, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True
-    ) -> dict[str, MainSchemaTypes]:
+    def get_full(self, branch: Branch | str | None = None, duplicate: bool = True) -> dict[str, MainSchemaTypes]:
         branch_name = get_branch_name(branch=branch)
         if branch_name not in self._db._schemas:
             return registry.schema.get_full(branch=branch)
         return self._db._schemas[branch_name].get_all(duplicate=duplicate)
 
     async def get_full_safe(
-        self, branch: Optional[Union[Branch, str]] = None, duplicate: bool = True
+        self, branch: Branch | str | None = None, duplicate: bool = True
     ) -> dict[str, MainSchemaTypes]:
         await lock.registry.local_schema_wait()
         return self.get_full(branch=branch, duplicate=duplicate)
@@ -206,10 +202,10 @@ def get_context(self) -> dict[str, Any]:
 
         return {}
 
-    def add_schema(self, schema: SchemaBranch, name: Optional[str] = None) -> None:
+    def add_schema(self, schema: SchemaBranch, name: str | None = None) -> None:
         self._schemas[name or schema.name] = schema
 
-    def start_session(self, read_only: bool = False, schemas: Optional[list[SchemaBranch]] = None) -> InfrahubDatabase:
+    def start_session(self, read_only: bool = False, schemas: list[SchemaBranch] | None = None) -> InfrahubDatabase:
         """Create a new InfrahubDatabase object in Session mode."""
         session_mode = InfrahubDatabaseSessionMode.WRITE
         if read_only:
@@ -229,7 +225,7 @@ def start_session(self, read_only: bool = False, schemas: Optional[list[SchemaBr
             **context,
         )
 
-    def start_transaction(self, schemas: Optional[list[SchemaBranch]] = None) -> InfrahubDatabase:
+    def start_transaction(self, schemas: list[SchemaBranch] | None = None) -> InfrahubDatabase:
         context = self.get_context()
 
         return self.__class__(
@@ -261,7 +257,7 @@ async def session(self) -> AsyncSession:
         self._is_session_local = True
         return self._session
 
-    async def transaction(self, name: Optional[str]) -> AsyncTransaction:
+    async def transaction(self, name: str | None) -> AsyncTransaction:
         if self._transaction:
             return self._transaction
 
@@ -290,9 +286,9 @@ async def __aenter__(self) -> 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,
     ):
         if self._mode == InfrahubDatabaseMode.SESSION:
             return await self._session.close()
@@ -385,9 +381,9 @@ async def execute_query_with_metadata(
                 return results, response._metadata or {}
 
     async def run_query(
-        self, query: str, params: Optional[dict[str, Any]] = None, name: Optional[str] = "undefined"
+        self, query: str, params: dict[str, Any] | None = None, name: str | None = "undefined"
     ) -> AsyncResult:
-        _query: Union[str | Query] = query
+        _query: str | Query = query
         if self.is_transaction:
             execution_method = await self.transaction(name=name)
         else:

backend/infrahub/git/base.py

@@ -3,7 +3,7 @@
 import shutil
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import TYPE_CHECKING, NoReturn, Optional, Union
+from typing import TYPE_CHECKING, NoReturn
 from uuid import UUID  # noqa: TC003
 
 import git
@@ -144,12 +144,12 @@ class InfrahubRepositoryBase(BaseModel, ABC):
         False, description="Flag to indicate if a remote repository (named origin) is present in the config."
     )
 
-    client: Optional[InfrahubClient] = Field(
+    client: InfrahubClient | None = Field(
         default=None,
         description="Infrahub Client, used to query the Repository and Branch information in the graph and to update the commit.",
     )
 
-    cache_repo: Optional[Repo] = Field(None, description="Internal cache of the GitPython Repo object")
+    cache_repo: Repo | None = Field(None, description="Internal cache of the GitPython Repo object")
     service: InfrahubServices = Field(
         ..., description="Service object with access to the message queue, the database etc.."
     )
@@ -584,7 +584,7 @@ async def create_branch_in_git(self, branch_name: str, branch_id: str | None = N
 
         return True
 
-    def create_commit_worktree(self, commit: str) -> Union[bool, Worktree]:
+    def create_commit_worktree(self, commit: str) -> bool | Worktree:
         """Create a new worktree for a given commit."""
 
         # Check of the worktree already exist
@@ -744,7 +744,7 @@ async def pull(
         branch_id: str | None = None,
         create_if_missing: bool = False,
         update_commit_value: bool = True,
-    ) -> Union[bool, str]:
+    ) -> bool | str:
         """Pull the latest update from the remote repository on a given branch."""
 
         if not self.has_origin:
@@ -812,10 +812,10 @@ async def get_conflicts(self, source_branch: str, dest_branch: str) -> list[str]
 
     async def find_files(
         self,
-        extension: Union[str, list[str]],
+        extension: str | list[str],
         branch_name: str | None = None,
         commit: str | None = None,
-        directory: Optional[Path] = None,
+        directory: Path | None = None,
     ) -> list[Path]:
         """Return the path of all files matching a specific extension in a given Branch or Commit."""
         if not branch_name and not commit:

backend/infrahub/git/integrator.py

@@ -3,7 +3,7 @@
 import hashlib
 import importlib
 import sys
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any
 
 import jinja2
 import ujson
@@ -93,10 +93,10 @@ class CheckDefinitionInformation(BaseModel):
     timeout: int
     """Timeout for the Check."""
 
-    parameters: Optional[dict] = None
+    parameters: dict | None = None
     """Additional Parameters to extract from each target (if targets is provided)"""
 
-    targets: Optional[str] = Field(default=None, description="Targets if not a global check")
+    targets: str | None = Field(default=None, description="Targets if not a global check")
 
 
 class TransformPythonInformation(BaseModel):
@@ -161,7 +161,7 @@ async def ensure_location_is_defined(self) -> None:
 
     @flow(name="import-object-from-file", flow_run_name="Import objects")
     async def import_objects_from_files(
-        self, infrahub_branch_name: str, git_branch_name: Optional[str] = None, commit: Optional[str] = None
+        self, infrahub_branch_name: str, git_branch_name: str | None = None, commit: str | None = None
     ) -> None:
         if not commit:
             commit = self.get_commit_value(branch_name=git_branch_name or infrahub_branch_name)
@@ -443,7 +443,7 @@ async def update_artifact_definition(
         await existing_artifact_definition.save()
 
     @task(name="repository-get-config", task_run_name="get repository config", cache_policy=NONE)  # type: ignore[arg-type]
-    async def get_repository_config(self, branch_name: str, commit: str) -> Optional[InfrahubRepositoryConfig]:
+    async def get_repository_config(self, branch_name: str, commit: str) -> InfrahubRepositoryConfig | None:
         branch_wt = self.get_worktree(identifier=commit or branch_name)
         log = get_run_logger()
 
@@ -1098,7 +1098,7 @@ async def execute_python_check(
         location: str,
         class_name: str,
         client: InfrahubClient,
-        params: Optional[dict] = None,
+        params: dict | None = None,
     ) -> InfrahubCheck:
         """Execute A Python Check stored in the repository."""
         log = get_run_logger()
@@ -1270,7 +1270,7 @@ async def render_artifact(
         self,
         artifact: CoreArtifact,
         artifact_created: bool,
-        message: Union[CheckArtifactCreate, RequestArtifactGenerate],
+        message: CheckArtifactCreate | RequestArtifactGenerate,
     ) -> ArtifactGenerateResult:
         response = await self.sdk.query_gql_query(
             name=message.query,

backend/infrahub/git/models.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from pydantic import BaseModel, ConfigDict, Field
 
 from infrahub.context import InfrahubContext
@@ -34,7 +32,7 @@ class RequestArtifactGenerate(BaseModel):
     target_id: str = Field(..., description="The ID of the target object for this artifact")
     target_kind: str = Field(..., description="The kind of the target object for this artifact")
     target_name: str = Field(..., description="Name of the artifact target")
-    artifact_id: Optional[str] = Field(default=None, description="The id of the artifact if it previously existed")
+    artifact_id: str | None = Field(default=None, description="The id of the artifact if it previously existed")
     query: str = Field(..., description="The name of the query to use when collecting data")
     timeout: int = Field(..., description="Timeout for requests used to generate this artifact")
     variables: dict = Field(..., description="Input variables when generating the artifact")
@@ -47,8 +45,8 @@ class GitRepositoryAdd(BaseModel):
     location: str = Field(..., description="The external URL of the repository")
     repository_id: str = Field(..., description="The unique ID of the Repository")
     repository_name: str = Field(..., description="The name of the repository")
-    created_by: Optional[str] = Field(default=None, description="The user ID of the user that created the repository")
-    default_branch_name: Optional[str] = Field(None, description="Default branch for this repository")
+    created_by: str | None = Field(default=None, description="The user ID of the user that created the repository")
+    default_branch_name: str | None = Field(None, description="Default branch for this repository")
     infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
     infrahub_branch_id: str = Field(..., description="Id of the Infrahub branch on which to sync the remote repository")
     internal_status: str = Field(..., description="Administrative status of the repository")
@@ -61,7 +59,7 @@ class GitRepositoryAddReadOnly(BaseModel):
     repository_id: str = Field(..., description="The unique ID of the Repository")
     repository_name: str = Field(..., description="The name of the repository")
     ref: str = Field(..., description="Ref to track on the external repository")
-    created_by: Optional[str] = Field(default=None, description="The user ID of the user that created the repository")
+    created_by: str | None = Field(default=None, description="The user ID of the user that created the repository")
     infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
     infrahub_branch_id: str = Field(..., description="Id of the Infrahub branch on which to sync the remote repository")
     internal_status: str = Field(..., description="Internal status of the repository")
@@ -73,8 +71,8 @@ class GitRepositoryPullReadOnly(BaseModel):
     location: str = Field(..., description="The external URL of the repository")
     repository_id: str = Field(..., description="The unique ID of the Repository")
     repository_name: str = Field(..., description="The name of the repository")
-    ref: Optional[str] = Field(None, description="Ref to track on the external repository")
-    commit: Optional[str] = Field(None, description="Specific commit to pull")
+    ref: str | None = Field(None, description="Ref to track on the external repository")
+    commit: str | None = Field(None, description="Specific commit to pull")
     infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")
     infrahub_branch_id: str = Field(..., description="Infrahub branch on which to sync the remote repository")
 
@@ -108,7 +106,7 @@ class GitDiffNamesOnly(BaseModel):
     repository_name: str = Field(..., description="The name of the repository")
     repository_kind: str = Field(..., description="The kind of the repository")
     first_commit: str = Field(..., description="The first commit")
-    second_commit: Optional[str] = Field(None, description="The second commit")
+    second_commit: str | None = Field(None, description="The second commit")
 
 
 class GitDiffNamesOnlyResponse(BaseModel):