Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions backend/infrahub/core/branch/models.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

import re
from typing import TYPE_CHECKING, Any, Optional, Self, Union, cast
from typing import TYPE_CHECKING, Any, Self, cast

from pydantic import Field, field_validator

Expand Down Expand Up @@ -34,7 +34,7 @@ class Branch(StandardNode):
status: BranchStatus = BranchStatus.OPEN
description: str = ""
origin_branch: str = "main"
branched_from: Optional[str] = Field(default=None, validate_default=True)
branched_from: str | None = Field(default=None, validate_default=True)
hierarchy_level: int = 2
is_default: bool = False
is_global: bool = False
Expand All @@ -44,8 +44,8 @@ class Branch(StandardNode):
description="Indicate if the branch should be extended to Git and if Infrahub should merge the branch in Git as part of a proposed change",
)
is_isolated: bool = True
schema_changed_at: Optional[str] = None
schema_hash: Optional[SchemaBranchHash] = None
schema_changed_at: str | None = None
schema_hash: SchemaBranchHash | None = None
graph_version: int | None = None

_exclude_attrs: list[str] = ["id", "uuid", "owner"]
Expand Down Expand Up @@ -220,7 +220,7 @@ async def get_list_count(
def isinstance(cls, obj: Any) -> bool:
return isinstance(obj, cls)

def get_origin_branch(self) -> Optional[Branch]:
def get_origin_branch(self) -> Branch | None:
"""Return the branch Object of the origin_branch."""
if not self.origin_branch or self.origin_branch == self.name:
return None
Expand All @@ -239,7 +239,7 @@ def get_branches_in_scope(self) -> list[str]:

return [default_branch, self.name]

def get_branches_and_times_to_query(self, at: Optional[Timestamp] = None) -> dict[frozenset, str]:
def get_branches_and_times_to_query(self, at: Timestamp | None = None) -> dict[frozenset, str]:
"""Return all the names of the branches that are constituing this branch with the associated times excluding the global branch"""

at = Timestamp(at)
Expand All @@ -260,7 +260,7 @@ def get_branches_and_times_to_query(self, at: Optional[Timestamp] = None) -> dic

def get_branches_and_times_to_query_global(
self,
at: Optional[Timestamp] = None,
at: Timestamp | None = None,
is_isolated: bool = True,
) -> dict[frozenset, str]:
"""Return all the names of the branches that are constituting this branch with the associated times."""
Expand Down Expand Up @@ -330,7 +330,7 @@ async def delete(self, db: InfrahubDatabase) -> None:
await super().delete(db=db)

def get_query_filter_relationships(
self, rel_labels: list, at: Optional[Timestamp] = None, include_outside_parentheses: bool = False
self, rel_labels: list, at: Timestamp | None = None, include_outside_parentheses: bool = False
) -> tuple[list, dict]:
"""
Generate a CYPHER Query filter based on a list of relationships to query a part of the graph at a specific time and on a specific branch.
Expand Down Expand Up @@ -368,7 +368,7 @@ def get_query_filter_relationships(

def get_query_filter_path(
self,
at: Optional[Union[Timestamp, str]] = None,
at: Timestamp | str | None = None,
is_isolated: bool = True,
branch_agnostic: bool = False,
variable_name: str = "r",
Expand Down
6 changes: 3 additions & 3 deletions backend/infrahub/core/graph/constraints.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

from enum import StrEnum
from typing import TYPE_CHECKING, ForwardRef, Optional, Union, get_origin
from typing import TYPE_CHECKING, ForwardRef, Union, get_origin

from pydantic import BaseModel
from typing_extensions import Self
Expand Down Expand Up @@ -162,8 +162,8 @@ def get_query_exist_drop(self) -> str:


class ConstraintManagerBase:
constraint_node_class: Optional[type[ConstraintItem]] = ConstraintItem
constraint_rel_class: Optional[type[ConstraintItem]] = ConstraintItem
constraint_node_class: type[ConstraintItem] | None = ConstraintItem
constraint_rel_class: type[ConstraintItem] | None = ConstraintItem

def __init__(self, db: InfrahubDatabase) -> None:
self.db = db
Expand Down
8 changes: 4 additions & 4 deletions backend/infrahub/core/graph/schema.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, Field

Expand Down Expand Up @@ -215,7 +215,7 @@ class GraphBooleanNode(GraphVertex):

class GraphRelationshipIsPartOf(BaseModel):
from_: str = Field(..., description="Time from which the relationship is valid", alias="from")
to_: Optional[str] = Field(None, description="Time until which the relationship is valid", alias="to")
to_: str | None = Field(None, description="Time until which the relationship is valid", alias="to")
status: RelationshipStatus = Field(..., description="status of the relationship")


Expand All @@ -229,9 +229,9 @@ class GraphRelationshipDefault(BaseModel):
ge=1,
)
from_: str = Field(..., description="Time from which the relationship is valid", alias="from")
to_: Optional[str] = Field(None, description="Time until which the relationship is valid", alias="to")
to_: str | None = Field(None, description="Time until which the relationship is valid", alias="to")
status: RelationshipStatus = Field(..., description="status of the relationship")
hierarchy: Optional[str] = Field(None, description="Name of the hierarchy this relationship is part of")
hierarchy: str | None = Field(None, description="Name of the hierarchy this relationship is part of")


def get_graph_schema() -> dict[str, dict[str, type[Any]]]:
Expand Down
23 changes: 12 additions & 11 deletions backend/infrahub/core/node/standard.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@
import inspect
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING, Any, Optional, Union, get_args, get_origin
from types import UnionType
from typing import TYPE_CHECKING, Any, Union, get_args, get_origin
from uuid import UUID

import ujson
Expand Down Expand Up @@ -45,12 +46,12 @@ class StandardNodeQueryFields:


class StandardNode(BaseModel):
id: Optional[str] = None
uuid: Optional[UUID] = None
created_at: Optional[str] = Field(default=None, validate_default=True)
id: str | None = None
uuid: UUID | None = None
created_at: str | None = Field(default=None, validate_default=True)
created_by: str = Field(default=SYSTEM_USER_ID)
updated_by: Optional[str] = Field(default=None)
updated_at: Optional[str] = Field(default=None, validate_default=True)
updated_by: str | None = Field(default=None)
updated_at: str | None = Field(default=None, validate_default=True)

_query: type[StandardNodeQuery] = StandardNodeCreateQuery
_exclude_attrs: list[str] = ["id", "uuid", "_query"]
Expand Down Expand Up @@ -79,9 +80,9 @@ def guess_field_type(field: FieldInfo) -> Any:
annotation_origin = get_origin(field.annotation)
annotation_args = get_args(field.annotation)

if (annotation_origin == Union and len(annotation_args) == 2 and type(None) in annotation_args) or (
annotation_origin is list and len(annotation_args)
):
if (
annotation_origin in (Union, UnionType) and len(annotation_args) == 2 and type(None) in annotation_args
) or (annotation_origin is list and len(annotation_args)):
return get_origin(annotation_args[0]) or annotation_args[0]

return annotation_origin or field.annotation
Expand Down Expand Up @@ -212,7 +213,7 @@ async def update(self, db: InfrahubDatabase, user_id: str = SYSTEM_USER_ID) -> b
return True

@classmethod
async def get(cls, id: str, db: InfrahubDatabase) -> Optional[Self]:
async def get(cls, id: str, db: InfrahubDatabase) -> Self | None:
"""Get a node from the database identified by its ID."""

node = await cls._get_item_raw(id=id, db=db)
Expand All @@ -233,7 +234,7 @@ async def _get_item_raw(cls, id: str, db: InfrahubDatabase) -> Neo4jNode:
return result.get("n")

@classmethod
def from_db(cls, node: Neo4jNode, extras: Optional[dict[str, Any]] = None) -> Self:
def from_db(cls, node: Neo4jNode, extras: dict[str, Any] | None = None) -> Self:
"""Convert a Neo4j Node to a Infrahub StandardNode

Args:
Expand Down
29 changes: 0 additions & 29 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -625,42 +625,13 @@ allow-dunder-method-names = [
"PLR0915", # Too many statements
]

"backend/infrahub/core/branch/models.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
##################################################################################################
"UP007", # Use X | Y for type annotations
"UP045", # Use X | Y for type annotations
]

"backend/infrahub/core/graph/**.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
##################################################################################################
"UP045", # Use X | Y for type annotations
]

"backend/infrahub/core/node/__init__.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
##################################################################################################
"RET503", # Missing explicit `return` at the end of function able to return non-`None` value
]

"backend/infrahub/core/node/standard.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
##################################################################################################
"UP045", # Use X | Y for type annotations
]

"backend/infrahub/core/query/**.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
##################################################################################################
"UP007", # Use X | Y for type annotations
]

"backend/infrahub/core/models.py" = [
##################################################################################################
# Refactor code and remove the ignore rule
Expand Down
Loading