[core][feat] Use view to query the graph (#2101)

Author: aquamatthias

fixcore/fixcore/cli/cli.py

@@ -216,6 +216,7 @@ def help_command() -> JsGen:
 
 CLIArg = Tuple[CLICommand, Optional[str]]
 # If no sort is defined in the part, we use this default sort order
+# Note: changing the default sort order should be reflected in the graphdb search view (fix_view)
 DefaultSort = [Sort("/reported.kind"), Sort("/reported.name"), Sort("/reported.id")]
 # Default sort order for history searches
 HistorySort = [Sort("/changed_at"), Sort("/reported.kind"), Sort("/reported.name"), Sort("/reported.id")]

fixcore/fixcore/cli/command.py

@@ -1464,37 +1464,35 @@ async def get_db(at: Optional[datetime], graph_name: GraphName) -> Tuple[GraphDB
 
             return db_access.get_graph_db(graph_name), graph_name
 
-        async def load_query_model(db: GraphDB, graph_name: GraphName) -> QueryModel:
+        async def load_query_model(graph_name: GraphName) -> QueryModel:
             model = await self.dependencies.model_handler.load_model(graph_name)
-            query_model = QueryModel(query, model, ctx.env)
-            await db.to_query(query_model)  # only here to validate the query itself (can throw)
-            return query_model
+            return QueryModel(query, model, ctx.env)
 
         async def explain_search() -> AsyncIterator[Json]:
             db, graph_name = await get_db(at, current_graph_name)
-            query_model = await load_query_model(db, graph_name)
+            query_model = await load_query_model(graph_name)
             explanation = await db.explain(query_model, with_edges)
             yield to_js(explanation)
 
         async def prepare() -> Tuple[CLISourceContext, AsyncIterator[Json]]:
             db, graph_name = await get_db(at, current_graph_name)
 
-            query_model = await load_query_model(db, graph_name)
+            query_model = await load_query_model(graph_name)
             count = ctx.env.get("count", "true").lower() != "false"
             timeout = if_set(ctx.env.get("search_timeout"), duration)
+            consistent = ctx.env.get("consistent")
+            search_args = dict(with_count=count, timeout=timeout, consistent=consistent)
             if history:
                 before = if_set(parsed.get("before"), lambda x: parse_time_or_delta(strip_quotes(x)))
                 after = if_set(parsed.get("after"), lambda x: parse_time_or_delta(strip_quotes(x)))
                 changes = [HistoryChange[strip_quotes(x)] for x in parsed.get("change", [])]
-                context = await db.search_history(
-                    query_model, changes, before, after, with_count=count, timeout=timeout
-                )
+                context = await db.search_history(query_model, changes, before, after, **search_args)
             elif query.aggregate:
-                context = await db.search_aggregation(query_model, with_count=count, timeout=timeout)
+                context = await db.search_aggregation(query_model, **search_args)
             elif with_edges:
-                context = await db.search_graph_gen(query_model, with_count=count, timeout=timeout)
+                context = await db.search_graph_gen(query_model, **search_args)
             else:
-                context = await db.search_list(query_model, with_count=count, timeout=timeout)
+                context = await db.search_list(query_model, **search_args)
             cursor = context.cursor
 
             # since we can not use context boundaries here,

fixcore/fixcore/core_config.py

@@ -496,7 +496,7 @@ def json(self) -> Json:
 
 
 @define()
-class GraphUpdateConfig(ConfigObject):
+class GraphConfig(ConfigObject):
     kind: ClassVar[str] = f"{FixCoreRoot}_graph_update_config"
     merge_max_wait_time_seconds: int = field(
         default=3600, metadata={"description": "Max waiting time to complete a merge graph action."}
@@ -517,6 +517,10 @@ class GraphUpdateConfig(ConfigObject):
         default=5,
         metadata={"description": "Number of parallel graph merge requests handled in parallel."},
     )
+    use_view: bool = field(
+        default=True,
+        metadata={"description": "If true, the graph uses an efficient view to query the data."},
+    )
 
     def merge_max_wait_time(self) -> timedelta:
         return timedelta(seconds=self.merge_max_wait_time_seconds)
@@ -527,7 +531,7 @@ def abort_after(self) -> timedelta:
 
 # Define rules to validate this config
 schema_registry.add(
-    schema_name(GraphUpdateConfig),
+    schema_name(GraphConfig),
     dict(
         merge_max_wait_time_seconds={"type": "integer", "min": 60},
         abort_after_seconds={"type": "integer", "min": 60},
@@ -609,7 +613,7 @@ class TimeSeriesConfig(ConfigObject):
 class CoreConfig(ConfigObject):
     api: ApiConfig
     cli: CLIConfig
-    graph_update: GraphUpdateConfig
+    graph: GraphConfig
     runtime: RuntimeConfig
     db: DatabaseConfig
     workflows: Dict[str, WorkflowConfig]
@@ -629,7 +633,7 @@ def no_scheduling(self) -> bool:
 
     @property
     def editable(self) -> "EditableConfig":
-        return EditableConfig(self.api, self.cli, self.graph_update, self.runtime, self.workflows, self.timeseries)
+        return EditableConfig(self.api, self.cli, self.graph, self.runtime, self.workflows, self.timeseries)
 
     def json(self) -> Json:
         return {FixCoreRoot: to_js(self.editable, strip_attr="kind")}
@@ -649,8 +653,8 @@ class EditableConfig(ConfigObject):
         factory=CLIConfig,
         metadata={"description": "CLI related properties."},
     )
-    graph_update: GraphUpdateConfig = field(
-        factory=GraphUpdateConfig,
+    graph: GraphConfig = field(
+        factory=GraphConfig,
         metadata={"description": "Properties for updating the graph."},
     )
     runtime: RuntimeConfig = field(
@@ -679,7 +683,7 @@ def config_model() -> List[Json]:
     dict(
         api={"schema": schema_name(ApiConfig), "allow_unknown": True},
         cli={"schema": schema_name(CLIConfig), "allow_unknown": True},
-        graph_update={"schema": schema_name(GraphUpdateConfig), "allow_unknown": True},
+        graph_update={"schema": schema_name(GraphConfig), "allow_unknown": True},
         runtime={"schema": schema_name(RuntimeConfig), "allow_unknown": True},
         workflows={
             "type": "dict",
@@ -774,7 +778,7 @@ def parse_config(
         custom_commands=commands_config,
         snapshots=snapshots_config,
         db=db,
-        graph_update=ed.graph_update,
+        graph=ed.graph,
         runtime=ed.runtime,
         workflows=ed.workflows,
         run=RunConfig(),  # overridden for each run
@@ -790,23 +794,12 @@ def migrate_core_config(config: Json) -> Json:
     cfg = config.get(FixCoreRoot) or {}
     adapted = deepcopy(cfg)
 
-    # 2.2 -> 2.3: rename and toggle `analytics_opt_out` -> `usage_metrics`
-    opt_out = value_in_path(cfg, "runtime.analytics_opt_out")
-    usage = value_in_path(cfg, "runtime.usage_metrics")
-    if opt_out is not None and usage is None:
-        set_value_in_path(not opt_out, "runtime.usage_metrics", adapted)
-    del_value_in_path(adapted, "runtime.analytics_opt_out")
-
-    # 3.0 -> 3.1: delete `api.ui_path`
-    del_value_in_path(adapted, "api.ui_path")
-
-    # 3.5 -> 3.6: web_port -> https_port
-    if web_port := value_in_path(cfg, "api.web_port"):
-        set_value_in_path(web_port, "api.https_port", adapted)
-        del_value_in_path(adapted, "api.web_port")
-
-    if value_in_path(cfg, "runtime.plantuml_server") == "http://plantuml.resoto.org:8080":
-        set_value_in_path("https://plantuml.fix.security", "runtime.plantuml_server", adapted)
+    # 4.0 -> 4.1: graph_update -> graph
+    if graph_update := value_in_path(cfg, "graph_update"):
+        log.info("Migrate fix.core config: rename graph_update to graph.")
+        set_value_in_path(graph_update, "graph", adapted)
+        set_value_in_path(True, "graph.use_view", adapted)
+        del_value_in_path(adapted, "graph_update")
 
     return {FixCoreRoot: adapted}