HHT-669: extracting kg updater and repository

app/clients/influxdb/influxdb_client_impl.py

@@ -4,9 +4,9 @@
 from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
 
 from app.clients.influxdb.influxdb_settings import InfluxDBSettings
-from app.clients.influxdb.metricstore_client import MetricStoreClient
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.metricstore_client import MetricStoreClient
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class InfluxDBClientImpl(MetricStoreClient):

app/clients/influxdb/mock_infuxdbclient.py

@@ -1,8 +1,8 @@
 from typing import Dict, List
 
-from app.clients.influxdb.metricstore_client import MetricStoreClient
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.metricstore_client import MetricStoreClient
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class MockInfluxDBClient(MetricStoreClient):

app/clients/influxdb/simple_result_parser.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict, List
 
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class SimpleResultParser(QueryResultParser):

app/clients/influxdb/test_simple_result_parser.py

@@ -4,7 +4,7 @@
 from dateutil.tz import tzutc
 
 from app.clients.influxdb.simple_result_parser import SimpleResultParser
-from app.core.metric_value import MetricValue
+from app.core.types import MetricValue
 
 
 class SimpleResultParserTest(TestCase):

app/clients/k8s/k8s_client.py

@@ -54,6 +54,7 @@ def find_resources_by_kind_and_identity(
         else:
             return []
 
+    # TODO remove
     def get_resource_name(self, node: Dict[str, Any]) -> str:
         for match in parse("$.metadata.name").find(node):
             return str(match.value)

app/clients/prometheus/mock_prometheus_client.py

@@ -1,8 +1,8 @@
 from typing import Dict, List
 
-from app.clients.influxdb.metricstore_client import MetricStoreClient
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.metricstore_client import MetricStoreClient
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class MockPrometheusClient(MetricStoreClient):

app/clients/prometheus/prometheus_client.py

@@ -2,10 +2,10 @@
 
 from aioprometheus_api_client import PrometheusConnect
 
-from app.clients.influxdb.metricstore_client import MetricStoreClient
-from app.clients.influxdb.query_result_parser import QueryResultParser
 from app.clients.prometheus.prometheus_client_settings import PrometheusClientSettings
-from app.core.metric_value import MetricValue
+from app.core.repository.metricstore_client import MetricStoreClient
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class PrometheusClient(MetricStoreClient):

app/clients/prometheus/prometheus_result_parser.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict, List
 
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class PrometheusResultParser(QueryResultParser):

app/clients/prometheus/test_prometheus_result_parser.py

@@ -3,7 +3,7 @@
 from unittest import TestCase
 
 from app.clients.prometheus.prometheus_result_parser import PrometheusResultParser
-from app.core.metric_value import MetricValue
+from app.core.types import MetricValue
 
 
 class PrometheusResultParserTest(TestCase):

app/core/builder/kg_builder.py

@@ -12,9 +12,10 @@
 from app.core.builder.slice_strategy.single_slice_strategy import SingleSliceStrategy
 from app.core.builder.slice_strategy.slice_for_node_strategy import SliceForNodeStrategy
 from app.core.builder.slice_strategy.slice_strategy import SliceStrategy
-from app.core.kg_repository import KGRepository
-from app.core.metric_repository import MetricQuery, MetricRepository
+from app.core.repository.metric_repository import MetricRepository
+from app.core.repository.types import MetricQuery
 from app.core.types import DKGSlice, MetricSnapshot
+from app.core.updater.kg_repository import KGRepository
 from app.util.clock import Clock
 
 

app/core/builder/test_kg_builder.py

@@ -9,11 +9,11 @@
 )
 from app.core.async_queue import AsyncQueue
 from app.core.builder.kg_builder import KGBuilder, KGBuilderSettings, QuerySettings
-from app.core.kg_repository import KGRepository
-from app.core.metric_repository import MetricRepository
+from app.core.repository.metric_repository import MetricRepository
 from app.core.test_graph_fixture import TestGraphFixture
 from app.core.test_snapshot_base import SnapshotTestBase
 from app.core.types import DKGSlice, KGSliceId
+from app.core.updater.kg_repository import KGRepository
 from app.kg.inmemory_graph import InMemoryGraph
 from app.util.clock import Clock
 from app.util.mock_clock import MockClock

app/core/repository/metric_repository.py

@@ -0,0 +1,25 @@
+from typing import List
+
+import asyncio
+
+from app.core.repository.metricstore_client import MetricStoreClient
+from app.core.repository.types import MetricQuery
+from app.core.types import MetricValue
+
+
+class MetricRepository:
+    client: MetricStoreClient
+
+    def __init__(self, client: MetricStoreClient):
+        self.client = client
+
+    async def query_many(
+        self, now: int, queries: List[MetricQuery]
+    ) -> List[MetricValue]:
+        query_futures = [self.query_one(now, query) for query in queries]
+        query_results: List[List[MetricValue]] = await asyncio.gather(*query_futures)
+        return [element for elements in query_results for element in elements]
+
+    async def query_one(self, now: int, query: MetricQuery) -> List[MetricValue]:
+        result_parser = query.result_parser.get_by_name()
+        return await self.client.query(query.query, result_parser)

app/clients/influxdb/metricstore_client.py → app/core/repository/metricstore_client.py

@@ -1,7 +1,7 @@
 from typing import List
 
-from app.clients.influxdb.query_result_parser import QueryResultParser
-from app.core.metric_value import MetricValue
+from app.core.repository.query_result_parser import QueryResultParser
+from app.core.types import MetricValue
 
 
 class MetricStoreClient:

app/clients/influxdb/query_result_parser.py → app/core/repository/query_result_parser.py

@@ -2,7 +2,7 @@
 
 from datetime import datetime
 
-from app.core.metric_value import MetricValue
+from app.core.types import MetricValue
 
 
 class QueryResultParser:

app/core/test_metric_repository.py → app/core/repository/test_metric_repository.py

@@ -2,8 +2,9 @@
 from unittest import TestCase
 
 from app.clients.influxdb.mock_infuxdbclient import MockInfluxDBClient
-from app.core.metric_repository import MetricQuery, MetricRepository, ResultParserId
-from app.core.metric_value import MetricValue
+from app.core.repository.metric_repository import MetricRepository
+from app.core.repository.types import MetricQuery, ResultParserId
+from app.core.types import MetricValue
 
 
 class MetricRepositoryTest(TestCase):

app/core/repository/types.py

@@ -0,0 +1,34 @@
+from typing import Any, Optional
+
+import importlib
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class ResultParserId(StrEnum):
+    SIMPLE_RESULT_PARSER = (
+        "app.clients.influxdb.simple_result_parser.SimpleResultParser"
+    )
+
+    def get_by_name(self) -> Any:
+        if self.name == "SIMPLE_RESULT_PARSER":
+            return self.instantiate(self.value)
+        else:
+            raise Exception(f"Unknown parser for {self}")
+
+    def instantiate(self, clazz: str) -> Any:
+        idx = clazz.rfind(".")
+        module, class_name = clazz[:idx], clazz[idx + 1 :]
+        ClassObj = getattr(importlib.import_module(module), class_name)
+        return ClassObj()
+
+
+@dataclass
+class MetricQuery:
+    measurement_id: str
+    subresource: Optional[str]
+    source: str
+    unit: str
+    property: str
+    query: str
+    result_parser: ResultParserId

app/core/test_snapshot_base.py

@@ -11,9 +11,8 @@
 from app.clients.k8s.k8s_client import ResourceSnapshot
 from app.clients.k8s.mock_k8s_client import MockK8SClient
 from app.core.builder.kg_builder import QuerySettings
-from app.core.metric_repository import MetricQuery, ResultParserId
-from app.core.metric_value import MetricValue
-from app.core.types import KGSliceId, MetricSnapshot, SliceInputs
+from app.core.repository.types import MetricQuery, ResultParserId
+from app.core.types import KGSliceId, MetricSnapshot, MetricValue, SliceInputs
 from app.kg.graph import Graph
 from app.kg.id_base import IdBase
 from app.serialize.jsonld_configuration import JsonLDConfiguration

app/core/types.py

@@ -1,13 +1,23 @@
-from typing import Any, Dict, List, Set, Tuple
+from typing import Any, Dict, List, Set, Tuple, TypeAlias
 
 import urllib.parse
 from dataclasses import dataclass, field
 
 from app.clients.k8s.k8s_client import ResourceSnapshot
-from app.core.metric_repository import MetricQuery
-from app.core.metric_value import MetricValue
+from app.core.repository.types import MetricQuery
 from app.kg.graph import Graph
 
+MetricId: TypeAlias = str
+ResourceId: TypeAlias = str
+
+
+@dataclass
+class MetricValue:
+    metric_id: MetricId
+    resource_id: ResourceId
+    timestamp: int
+    value: float
+
 
 @dataclass(frozen=True)
 class KGSliceId:

app/core/kg_repository.py → app/core/updater/kg_repository.py

@@ -3,8 +3,8 @@
 from io import StringIO
 
 from app.clients.metadata_service.metadata_service_client import MetadataServiceClient
-from app.core.kg_result_parser import KGResultParser
 from app.core.types import KGSliceId
+from app.core.updater.kg_result_parser import KGResultParser
 from app.kg.graph import Graph
 from app.serialize.jsonld_configuration import JsonLDConfiguration
 from app.serialize.jsonld_serializer import JsonLDSerialializer

app/core/kg_tuple_parser.py → app/core/updater/kg_tuple_parser.py

@@ -1,7 +1,7 @@
 from typing import List
 
 from app.clients.metadata_service.metadata_service_client import Triple
-from app.core.kg_result_parser import KGResultParser
+from app.core.updater.kg_result_parser import KGResultParser
 from app.kg.graph import Graph
 from app.kg.inmemory_graph import InMemoryGraph
 

app/core/kg_updater.py → app/core/updater/kg_updater.py

@@ -5,8 +5,8 @@
 
 from app.core.async_queue import AsyncQueue
 from app.core.dkg_slice_store import DKGSliceStore
-from app.core.kg_repository import KGRepository
 from app.core.types import DKGSlice
+from app.core.updater.kg_repository import KGRepository
 
 
 class KGUpdater:

app/core/test_kg_repository.py → app/core/updater/test_kg_repository.py

@@ -4,10 +4,10 @@
 from app.clients.metadata_service.mock_metadata_service_client import (
     MockMetadataServiceClient,
 )
-from app.core.kg_repository import KGRepository
-from app.core.kg_tuple_parser import KGTupleParser
 from app.core.test_graph_fixture import TestGraphFixture
 from app.core.types import KGSliceId
+from app.core.updater.kg_repository import KGRepository
+from app.core.updater.kg_tuple_parser import KGTupleParser
 from app.kg.inmemory_graph import InMemoryGraph
 
 

app/core/test_kg_tuple_parser.py → app/core/updater/test_kg_tuple_parser.py

@@ -1,6 +1,6 @@
 from unittest import TestCase
 
-from app.core.kg_tuple_parser import KGTupleParser
+from app.core.updater.kg_tuple_parser import KGTupleParser
 from app.kg.inmemory_graph import InMemoryGraph
 
 

app/core/test_kg_updater.py → app/core/updater/test_kg_updater.py

@@ -7,10 +7,10 @@
     SerializedGraph,
 )
 from app.core.async_queue import AsyncQueue
-from app.core.kg_repository import KGRepository
-from app.core.kg_updater import KGUpdater
 from app.core.test_graph_fixture import TestGraphFixture
 from app.core.types import DKGSlice, KGSliceId
+from app.core.updater.kg_repository import KGRepository
+from app.core.updater.kg_updater import KGUpdater
 
 
 class KGUpdaterTest(TestCase, TestGraphFixture):