Prevent invalid related nodes from being passed to the Database

nodestream/cli/operations/initialize_logger.py

@@ -4,6 +4,7 @@
 
 from pythonjsonlogger.jsonlogger import JsonFormatter
 
+from ...pipeline.meta import get_context
 from ..commands.nodestream_command import NodestreamCommand
 from .operation import Operation
 
@@ -13,8 +14,19 @@ def _get_logger_level():
 
 
 def configure_logging_with_json_defaults():
-    logging.basicConfig(level=_get_logger_level())
-    formatter = JsonFormatter("%(name)s %(levelname)s %(message)s", timestamp=True)
+    logging.basicConfig(level=_get_logger_level(), force=True)
+    old_record_factory = logging.getLogRecordFactory()
+
+    def record_factory(*args, **kwargs):
+        record = old_record_factory(*args, **kwargs)
+        record.pipeline_name = get_context().name
+        return record
+
+    logging.setLogRecordFactory(record_factory)
+
+    formatter = JsonFormatter(
+        "%(name)s %(levelname)s %(pipeline_name)s %(message)s", timestamp=True
+    )
     logger = logging.getLogger()  # Configure the root logger.
     logger.handlers[0].setFormatter(formatter)
 

nodestream/model/desired_ingestion.py

@@ -74,6 +74,13 @@ def add_relationship(
         outbound: bool,
         match_strategy: MatchStrategy,
     ):
+        if not related_node.is_valid:
+            LOGGER.warning(
+                "Identity value for related node was null. Skipping.",
+                extra=asdict(related_node),
+            )
+            return
+
         from_node, to_node = (
             (self.source, related_node) if outbound else (related_node, self.source)
         )

tests/unit/cli/operations/test_initialize_logger.py

@@ -1,6 +1,13 @@
+import logging
+
 import pytest
+from hamcrest import assert_that, contains_string
 
-from nodestream.cli.operations import InitializeLogger
+from nodestream.cli.operations.initialize_logger import (
+    InitializeLogger,
+    configure_logging_with_json_defaults,
+)
+from nodestream.pipeline.meta import start_context
 
 
 @pytest.mark.asyncio
@@ -23,3 +30,14 @@ async def test_initialize_logger_json_logging_unset(mocker):
     )
     await subject.perform(command)
     patch.assert_not_called()
+
+
+def test_logs_pipeline_name(capsys):
+    configure_logging_with_json_defaults()
+    logger = logging.getLogger("some")
+    pipeline_name = "test_pipeline_name"
+    with start_context(pipeline_name):
+        logger.info("some message")
+
+    captured_out = capsys.readouterr().err
+    assert_that(captured_out, contains_string(pipeline_name))

tests/unit/model/test_desired_ingestion.py

@@ -0,0 +1,45 @@
+import pytest
+
+from nodestream.model import DesiredIngestion, MatchStrategy, Node, Relationship
+
+
+@pytest.fixture
+def desired_ingestion():
+    return DesiredIngestion()
+
+
+@pytest.fixture
+def valid_node():
+    return Node("Foo", {"bar": "baz"})
+
+
+@pytest.fixture
+def invalid_node():
+    return Node()
+
+
+@pytest.fixture
+def valid_relationship():
+    return Relationship("IS_RELATED_TO")
+
+
+def test_add_relationship_valid_node(desired_ingestion, valid_node, valid_relationship):
+    desired_ingestion.add_relationship(
+        related_node=valid_node,
+        relationship=valid_relationship,
+        outbound=True,
+        match_strategy=MatchStrategy.EAGER,
+    )
+    assert len(desired_ingestion.relationships) == 1
+
+
+def test_add_relationship_invalid_node(
+    desired_ingestion, invalid_node, valid_relationship
+):
+    desired_ingestion.add_relationship(
+        related_node=invalid_node,
+        relationship=valid_relationship,
+        outbound=True,
+        match_strategy=MatchStrategy.EAGER,
+    )
+    assert len(desired_ingestion.relationships) == 0