Merge pull request #6404 from opsmill/pog-stable-to-develop-20250505

Merge stable into develop with resolved merge conflicts

Author: ogenstad

.gitignore

@@ -17,6 +17,9 @@ development/docker-compose.dev-override.yml
 **/.idea/**
 dist/*
 
+# As we might run collect command from this repo
+support_logs_*
+
 # Direnv files (https://direnv.net/)
 .direnv/
 .envrc

CHANGELOG.md

@@ -11,6 +11,22 @@ This project uses [*towncrier*](https://towncrier.readthedocs.io/) and the chang
 
 <!-- towncrier release notes start -->
 
+## [Infrahub - v1.2.8](https://github.com/opsmill/infrahub/tree/infrahub-v1.2.8) - 2025-05-01
+
+### Added
+
+- Added support for "convert_query_response" for Python transforms. The feature works the same was as with Generators. Note any non default branch will need to be rebased after this upgrade. ([#6383](https://github.com/opsmill/infrahub/issues/6383))
+- Enable HCL syntax highlighting for artifacts
+
+### Fixed
+
+- Improve performance when retrieving nodes that have thousands of relationships
+- Improve performance of git credential helper
+
+### Housekeeping
+
+- Background performance improvements due to Prefect 3.3.7 upgrade
+
 ## [Infrahub - v1.2.7](https://github.com/opsmill/infrahub/tree/infrahub-v1.2.7) - 2025-04-28
 
 ### Security

backend/infrahub/api/transformation.py

@@ -88,6 +88,7 @@ async def transform_python(
         branch=branch_params.branch.name,
         transform_location=f"{transform.file_path.value}::{transform.class_name.value}",
         timeout=transform.timeout.value,
+        convert_query_response=transform.convert_query_response.value or False,
         data=data,
     )
 

backend/infrahub/artifacts/models.py

@@ -12,6 +12,10 @@ class CheckArtifactCreate(BaseModel):
     content_type: str = Field(..., description="Content type of the artifact")
     transform_type: str = Field(..., description="The type of transform associated with this artifact")
     transform_location: str = Field(..., description="The transforms location within the repository")
+    convert_query_response: bool = Field(
+        default=False,
+        description="Indicate if the query response should be converted to InfrahubNode objects for Python transforms",
+    )
     repository_id: str = Field(..., description="The unique ID of the Repository")
     repository_name: str = Field(..., description="The name of the Repository")
     repository_kind: str = Field(..., description="The kind of the Repository")

backend/infrahub/cli/db.py

@@ -23,7 +23,7 @@
 from infrahub.core import registry
 from infrahub.core.graph import GRAPH_VERSION
 from infrahub.core.graph.constraints import ConstraintManagerBase, ConstraintManagerMemgraph, ConstraintManagerNeo4j
-from infrahub.core.graph.index import node_indexes, rel_indexes
+from infrahub.core.graph.index import attr_value_index, node_indexes, rel_indexes
 from infrahub.core.graph.schema import (
     GRAPH_SCHEMA,
     GraphAttributeProperties,
@@ -48,6 +48,8 @@
 from infrahub.core.validators.models.validate_migration import SchemaValidateMigrationData
 from infrahub.core.validators.tasks import schema_validate_migrations
 from infrahub.database import DatabaseType
+from infrahub.database.memgraph import IndexManagerMemgraph
+from infrahub.database.neo4j import IndexManagerNeo4j
 from infrahub.log import get_logger
 from infrahub.services import InfrahubServices
 from infrahub.services.adapters.message_bus.local import BusSimulator
@@ -59,6 +61,7 @@
 if TYPE_CHECKING:
     from infrahub.cli.context import CliContext
     from infrahub.database import InfrahubDatabase
+    from infrahub.database.index import IndexManagerBase
 
 app = AsyncTyper()
 app.add_typer(patch_app, name="patch")
@@ -249,14 +252,20 @@ async def index(
 
     context: CliContext = ctx.obj
     dbdriver = await context.init_db(retry=1)
-    dbdriver.manager.index.init(nodes=node_indexes, rels=rel_indexes)
+    if dbdriver.db_type is DatabaseType.MEMGRAPH:
+        index_manager: IndexManagerBase = IndexManagerMemgraph(db=dbdriver)
+    index_manager = IndexManagerNeo4j(db=dbdriver)
+
+    if config.SETTINGS.experimental_features.value_db_index:
+        node_indexes.append(attr_value_index)
+    index_manager.init(nodes=node_indexes, rels=rel_indexes)
 
     if action == IndexAction.ADD:
-        await dbdriver.manager.index.add()
+        await index_manager.add()
     elif action == IndexAction.DROP:
-        await dbdriver.manager.index.drop()
+        await index_manager.drop()
 
-    indexes = await dbdriver.manager.index.list()
+    indexes = await index_manager.list()
 
     console = Console()
 
@@ -412,7 +421,7 @@ async def update_core_schema(
             update_db=True,
         )
         default_branch.update_schema_hash()
-        rprint("The Core Schema has been updated")
+        rprint("The Core Schema has been updated, make sure to rebase any open branches after the upgrade")
         if debug:
             rprint(f"New schema hash: {default_branch.active_schema_hash.main}")
         await default_branch.save(db=dbt)

backend/infrahub/computed_attribute/tasks.py

@@ -113,6 +113,7 @@ async def process_transform(
             location=f"{transform.file_path.value}::{transform.class_name.value}",
             data=data,
             client=service.client,
+            convert_query_response=transform.convert_query_response.value,
         )  # type: ignore[misc]
 
         await service.client.execute_graphql(
@@ -301,15 +302,24 @@ async def computed_attribute_setup_jinja2(
 
         triggers = await gather_trigger_computed_attribute_jinja2()
 
-        for trigger in triggers:
-            if event_name != BranchDeletedEvent.event_name and trigger.branch == branch_name:
+        # Since we can have multiple trigger per NodeKind
+        # we need to extract the list of unique node that should be processed
+        # also
+        # Because the automation in Prefect doesn't capture all information about the computed attribute
+        # we can't tell right now if a given computed attribute has changed and need to be updated
+        unique_nodes: set[tuple[str, str, str]] = {
+            (trigger.branch, trigger.computed_attribute.kind, trigger.computed_attribute.attribute.name)
+            for trigger in triggers
+        }
+        for branch, kind, attribute_name in unique_nodes:
+            if event_name != BranchDeletedEvent.event_name and branch == branch_name:
                 await service.workflow.submit_workflow(
                     workflow=TRIGGER_UPDATE_JINJA_COMPUTED_ATTRIBUTES,
                     context=context,
                     parameters={
-                        "branch_name": trigger.branch,
-                        "computed_attribute_name": trigger.computed_attribute.attribute.name,
-                        "computed_attribute_kind": trigger.computed_attribute.kind,
+                        "branch_name": branch,
+                        "computed_attribute_name": attribute_name,
+                        "computed_attribute_kind": kind,
                     },
                 )
 
@@ -319,6 +329,7 @@ async def computed_attribute_setup_jinja2(
                 client=prefect_client,
                 triggers=triggers,
                 trigger_type=TriggerType.COMPUTED_ATTR_JINJA2,
+                force_update=False,
             )  # type: ignore[misc]
 
         log.info(f"{len(triggers)} Computed Attribute for Jinja2 automation configuration completed")
@@ -346,18 +357,29 @@ async def computed_attribute_setup_python(
 
         triggers_python, triggers_python_query = await gather_trigger_computed_attribute_python(db=db)
 
-        for trigger in triggers_python:
-            if event_name != BranchDeletedEvent.event_name and trigger.branch == branch_name:
-                log.info(
-                    f"Triggering update for {trigger.computed_attribute.computed_attribute.attribute.name} on {branch_name}"
-                )
+        # Since we can have multiple trigger per NodeKind
+        # we need to extract the list of unique node that should be processed
+        # also
+        # Because the automation in Prefect doesn't capture all information about the computed attribute
+        # we can't tell right now if a given computed attribute has changed and need to be updated
+        unique_nodes: set[tuple[str, str, str]] = {
+            (
+                trigger.branch,
+                trigger.computed_attribute.computed_attribute.kind,
+                trigger.computed_attribute.computed_attribute.attribute.name,
+            )
+            for trigger in triggers_python
+        }
+        for branch, kind, attribute_name in unique_nodes:
+            if event_name != BranchDeletedEvent.event_name and branch == branch_name:
+                log.info(f"Triggering update for {kind}.{attribute_name} on {branch}")
                 await service.workflow.submit_workflow(
                     workflow=TRIGGER_UPDATE_PYTHON_COMPUTED_ATTRIBUTES,
                     context=context,
                     parameters={
                         "branch_name": branch_name,
-                        "computed_attribute_name": trigger.computed_attribute.computed_attribute.attribute.name,
-                        "computed_attribute_kind": trigger.computed_attribute.computed_attribute.kind,
+                        "computed_attribute_name": attribute_name,
+                        "computed_attribute_kind": kind,
                     },
                 )
 

backend/infrahub/core/initialization.py

@@ -1,7 +1,9 @@
 import importlib
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from infrahub import config, lock
+from infrahub.constants.database import DatabaseType
 from infrahub.core import registry
 from infrahub.core.branch import Branch
 from infrahub.core.constants import (
@@ -13,6 +15,7 @@
     PermissionDecision,
 )
 from infrahub.core.graph import GRAPH_VERSION
+from infrahub.core.graph.index import attr_value_index, node_indexes, rel_indexes
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
 from infrahub.core.node.ipam import BuiltinIPPrefix
@@ -25,13 +28,18 @@
 from infrahub.core.schema import SchemaRoot, core_models, internal_schema
 from infrahub.core.schema.manager import SchemaManager
 from infrahub.database import InfrahubDatabase
+from infrahub.database.memgraph import IndexManagerMemgraph
+from infrahub.database.neo4j import IndexManagerNeo4j
 from infrahub.exceptions import DatabaseError
 from infrahub.graphql.manager import GraphQLSchemaManager
 from infrahub.log import get_logger
 from infrahub.menu.utils import create_default_menu
 from infrahub.permissions import PermissionBackend
 from infrahub.storage import InfrahubObjectStorage
 
+if TYPE_CHECKING:
+    from infrahub.database.index import IndexManagerBase
+
 log = get_logger()
 
 
@@ -115,7 +123,19 @@ async def initialize_registry(db: InfrahubDatabase, initialize: bool = False) ->
     registry.permission_backends = initialize_permission_backends()
 
 
-async def initialization(db: InfrahubDatabase) -> None:
+async def add_indexes(db: InfrahubDatabase) -> None:
+    if db.db_type is DatabaseType.MEMGRAPH:
+        index_manager: IndexManagerBase = IndexManagerMemgraph(db=db)
+    index_manager = IndexManagerNeo4j(db=db)
+
+    if config.SETTINGS.experimental_features.value_db_index:
+        node_indexes.append(attr_value_index)
+    index_manager.init(nodes=node_indexes, rels=rel_indexes)
+    log.debug("Loading database indexes ..")
+    await index_manager.add()
+
+
+async def initialization(db: InfrahubDatabase, add_database_indexes: bool = False) -> None:
     if config.SETTINGS.database.db_type == config.DatabaseType.MEMGRAPH:
         session = await db.session()
         await session.run(query="SET DATABASE SETTING 'log.level' TO 'INFO'")
@@ -129,12 +149,8 @@ async def initialization(db: InfrahubDatabase) -> None:
         log.debug("Checking Root Node")
         await initialize_registry(db=db, initialize=True)
 
-        # Add Indexes to the database
-        if db.manager.index.initialized:
-            log.debug("Loading database indexes ..")
-            await db.manager.index.add()
-        else:
-            log.warning("The database index manager hasn't been initialized.")
+        if add_database_indexes:
+            await add_indexes(db=db)
 
     # ---------------------------------------------------
     # Load all schema in the database into the registry

backend/infrahub/core/manager.py

@@ -1229,20 +1229,31 @@ async def _enrich_node_dicts_with_relationships(
         if not prefetch_relationships and not fields:
             return
         cardinality_one_identifiers_by_kind: dict[str, dict[str, RelationshipDirection]] | None = None
-        all_identifiers: list[str] | None = None
+        outbound_identifiers: set[str] | None = None
+        inbound_identifiers: set[str] | None = None
+        bidirectional_identifiers: set[str] | None = None
         if not prefetch_relationships:
             cardinality_one_identifiers_by_kind = _get_cardinality_one_identifiers_by_kind(
                 nodes=nodes_by_id.values(), fields=fields or {}
             )
-            all_identifiers_set: set[str] = set()
+            outbound_identifiers = set()
+            inbound_identifiers = set()
+            bidirectional_identifiers = set()
             for identifier_direction_map in cardinality_one_identifiers_by_kind.values():
-                all_identifiers_set.update(identifier_direction_map.keys())
-            all_identifiers = list(all_identifiers_set)
+                for identifier, direction in identifier_direction_map.items():
+                    if direction is RelationshipDirection.OUTBOUND:
+                        outbound_identifiers.add(identifier)
+                    elif direction is RelationshipDirection.INBOUND:
+                        inbound_identifiers.add(identifier)
+                    elif direction is RelationshipDirection.BIDIR:
+                        bidirectional_identifiers.add(identifier)
 
         query = await NodeListGetRelationshipsQuery.init(
             db=db,
             ids=list(nodes_by_id.keys()),
-            relationship_identifiers=all_identifiers,
+            outbound_identifiers=None if outbound_identifiers is None else list(outbound_identifiers),
+            inbound_identifiers=None if inbound_identifiers is None else list(inbound_identifiers),
+            bidirectional_identifiers=None if bidirectional_identifiers is None else list(bidirectional_identifiers),
             branch=branch,
             at=at,
             branch_agnostic=branch_agnostic,

backend/infrahub/core/migrations/graph/m014_remove_index_attr_value.py

@@ -7,6 +7,7 @@
 from infrahub.core.query import Query  # noqa: TC001
 from infrahub.database import DatabaseType
 from infrahub.database.index import IndexItem
+from infrahub.database.neo4j import IndexManagerNeo4j
 
 from ..shared import GraphMigration
 
@@ -30,8 +31,9 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:
             return result
 
         try:
-            db.manager.index.init(nodes=[INDEX_TO_DELETE], rels=[])
-            await db.manager.index.drop()
+            index_manager = IndexManagerNeo4j(db=db)
+            index_manager.init(nodes=[INDEX_TO_DELETE], rels=[])
+            await index_manager.drop()
         except Exception as exc:
             result.errors.append(str(exc))
             return result

backend/infrahub/core/protocols.py

@@ -478,6 +478,7 @@ class CoreTransformJinja2(CoreTransformation):
 class CoreTransformPython(CoreTransformation):
     file_path: String
     class_name: String
+    convert_query_response: BooleanOptional
 
 
 class CoreUserValidator(CoreValidator):