Merge pull request #395 from nodestream-proj/optional-timestamp-update

Flag to Disable last_ingested_at update

Author: zprobst

nodestream/databases/ingest_strategy.py

@@ -24,7 +24,7 @@ class IngestionStrategy(ABC):
     How that data is committed and how it is stored internally is decoupled through the `IngestionStrategy` interface.
 
     Generally, your usage of nodestream is decoupled from `IngestionStrategy` unless you intend to provide an implementation
-    of your own database writer. The writer API will give you a `DesiredIngestion` or other `Ingestible` object that needs a
+    of your own database writer. The writer API will give you a `DesiredIngestion` or other `Ingestible` object that needs an
     instance of an `IngestionStrategy` to apply operations to.
     """
 

nodestream/interpreting/interpretation_passes.py

@@ -63,7 +63,7 @@ def from_file_data(cls, interpretation_arg_list):
     def __init__(self, *interpretations: Interpretation):
         self.interpretations = interpretations
 
-    # Verifies that there is only one source node creator in all of the interpretations
+    # Verifies that there is only one source node creator in all the interpretations
     def verify_uniqueness(self):
         source_node_generator_count = 0
         for interpretation in self.interpretations:

nodestream/interpreting/interpretations/property_mapping.py

@@ -69,6 +69,16 @@ def __init__(self, map_of_value_providers: Dict[str, ValueProvider]):
     def key_value_generator(
         self, context: "ProviderContext", norm_args: Dict[str, bool]
     ):
+        """
+        Generates key-value pairs by normalizing values provided by ValueProviders.
+
+        Args:
+            context (ProviderContext): The context in which the values are being interpreted.
+            norm_args (Dict[str, bool]): Arguments used for normalizing the values.
+
+        Yields:
+            Tuple[str, Any]: A tuple containing the property key and its normalized value.
+        """
         for key, provider in self.map_of_value_providers.items():
             v = provider.normalize_single_value(context, norm_args)
             yield key, v
@@ -79,6 +89,14 @@ def apply_to(
         property_set: PropertySet,
         norm_args: Dict[str, bool],
     ):
+        """
+        Applies the key-value pairs generated by the key_value_generator to the given PropertySet.
+
+        Args:
+            context (ProviderContext): The context in which the values are being interpreted.
+            property_set (PropertySet): The PropertySet to which the key-value pairs will be applied.
+            norm_args (Dict[str, bool]): Arguments used for normalizing the values.
+        """
         property_set.apply(self.key_value_generator(context, norm_args))
 
     def __iter__(self):

nodestream/interpreting/interpretations/relationship_interpretation.py

@@ -94,6 +94,8 @@ class RelationshipInterpretation(Interpretation, alias="relationship"):
         "properties_normalization",
         "key_search_algorithm",
         "node_additional_types",
+        "node_update_last_ingested",
+        "relationship_update_last_ingested",
     )
 
     @deprecated_arugment("match_strategy", "node_creation_rule")
@@ -113,6 +115,8 @@ def __init__(
         key_normalization: Optional[Dict[str, Any]] = None,
         properties_normalization: Optional[Dict[str, Any]] = None,
         node_additional_types: Optional[Iterable[str]] = None,
+        node_update_last_ingested: bool = True,
+        relationship_update_last_ingested: bool = True,
     ):
         self.can_find_many = find_many or iterate_on is not None
         self.cardinality = Cardinality(cardinality)
@@ -146,6 +150,8 @@ def __init__(
             self.node_key, self.key_normalization
         )
         self.node_additional_types = tuple(node_additional_types or tuple())
+        self.node_update_last_ingested = node_update_last_ingested
+        self.relationship_update_last_ingested = relationship_update_last_ingested
 
     def interpret(self, context: ProviderContext):
         for sub_context in self.decomposer.decompose_record(context):
@@ -156,6 +162,10 @@ def interpret(self, context: ProviderContext):
 
     def find_relationship(self, context: ProviderContext) -> Relationship:
         rel = Relationship(type=self.relationship_type.single_value(context))
+
+        if not self.relationship_update_last_ingested:
+            rel.properties.remove_last_ingested()
+
         relationship_key_property_mapping = PropertyMappingFromDict(
             self.relationship_key
         )
@@ -174,6 +184,10 @@ def find_related_nodes(self, context: ProviderContext) -> Iterable[Node]:
                 key_values=PropertySet(key_set),
                 additional_types=self.node_additional_types,
             )
+
+            if not self.node_update_last_ingested:
+                node.properties.remove_last_ingested()
+
             if node.has_valid_id:
                 self.node_properties.apply_to(
                     context, node.properties, self.properties_normalization

nodestream/interpreting/interpretations/source_node_interpretation.py

@@ -83,6 +83,7 @@ class SourceNodeInterpretation(Interpretation, alias="source_node"):
         "additional_types",
         "norm_args",
         "allow_create",
+        "update_last_ingested",
     )
 
     def __init__(
@@ -96,6 +97,7 @@ def __init__(
         properties_normalization: Optional[Dict[str, Any]] = None,
         key_normalization: Optional[Dict[str, Any]] = None,
         allow_create: bool = True,
+        update_last_ingested: bool = True,
     ):
         if normalization and (properties_normalization or key_normalization):
             raise ValueError(
@@ -118,6 +120,7 @@ def __init__(
             self.creation_rule = NodeCreationRule.EAGER
         else:
             self.creation_rule = NodeCreationRule.MATCH_ONLY
+        self.update_last_ingested = update_last_ingested
 
     def interpret(self, context: ProviderContext):
         normalized_key: PropertySet = PropertySet()
@@ -135,6 +138,9 @@ def interpret(self, context: ProviderContext):
             normalized_properties,
         )
 
+        if not self.update_last_ingested:
+            context.desired_ingest.source.properties.remove_last_ingested()
+
     def expand_source_node_schema(self, source_node_schema: GraphObjectSchema):
         source_node_schema.add_keys(self.key)
         source_node_schema.add_properties(self.properties)

nodestream/model/graph_objects.py

@@ -86,10 +86,13 @@ def merge(self, properties: "PropertySet"):
         for key, val in properties.items():
             self.set_property(key, val)
 
+    def remove_last_ingested(self):
+        self.pop("last_ingested_at", None)
+
 
 @dataclass(slots=True)
 class Node(DeduplicatableObject):
-    """A `Node` is a entity that has a distinct identity.
+    """A `Node` is an entity that has a distinct identity.
 
     Each `Node` represents an entity (a person, place, thing, category or other piece of data) that has a distinct
     identity. Nodestream assumes the underlying graph database layer is a Labeled Property Graph. The identity

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nodestream"
-version = "0.13.7"
+version = "0.13.8"
 description = "A Fast, Declarative ETL for Graph Databases."
 license = "GPL-3.0-only"
 authors = [

tests/unit/interpreting/interpretations/test_relationship_interpretation.py

@@ -119,6 +119,8 @@ def test_relationship_interpretation_find_relationship_property_normalization(
         node_key={"hello": "world"},
         relationship_properties=supplied_properties,
         properties_normalization={"do_lowercase_strings": True},
+        node_update_last_ingested=False,
+        relationship_update_last_ingested=False,
     ).interpret(blank_context)
     assert_that(
         blank_context.desired_ingest.relationships[0].relationship.properties,

tests/unit/interpreting/interpretations/test_source_node_interpretation.py

@@ -66,6 +66,7 @@ def test_source_node_interpretation_applies_dynamic_properties(blank_context):
         node_type=EXPECTED_NODE_TYPE,
         key={},
         properties={"first_name": dynamic_first_name, "last_name": "Probst"},
+        update_last_ingested=False,
     )
     subject.interpret(blank_context)
     actual_properties = blank_context.desired_ingest.source.properties