OAM-240 Added price_changes resource (#13)

sigeca_data_export_microservice/app/application/synchronizations/resources_synchronization.py

@@ -5,7 +5,7 @@
     ProofOfDeliveryLineItemResourceReader, RequisitionResourceReader, RequisitionLineItemResourceReader, \
     StockCardResourceReader, StockCardLineItemResourceReader, StockEventResourceReader, \
     StockEventLineItemResourceReader, CalculatedStockOnHandResourceReader, SupportedProgramResourceReader, \
-    UserResourceReader
+    UserResourceReader, PriceChangesResourceReader
 
 
 class ChangeLogResourceSynchronization(ResourceSynchronization):
@@ -40,6 +40,10 @@ class FacilityResourceSynchronization(ResourceSynchronization):
     synchronized_resource = FacilityResourceReader
 
 
+class PriceChangesResourceSynchronization(ResourceSynchronization):
+    synchronized_resource = PriceChangesResourceReader
+
+
 class ProgramResourceSynchronization(ResourceSynchronization):
     synchronized_resource = ProgramResourceReader
 
@@ -106,6 +110,8 @@ def get_full_sync_list(jdbc_reader, api_client):
         OrderableResourceSynchronization(jdbc_reader, api_client),
         ProgramResourceSynchronization(jdbc_reader, api_client),
         ProgramOrderableResourceSynchronization(jdbc_reader, api_client),
+        # Has to be after ProgramOrderable
+        PriceChangesResourceSynchronization(jdbc_reader, api_client),
         ProofOfDeliveryResourceSynchronization(jdbc_reader, api_client),
         ProofOfDeliveryLineItemResourceSynchronization(jdbc_reader, api_client),
         RequisitionResourceSynchronization(jdbc_reader, api_client),

sigeca_data_export_microservice/app/domain/resources/__init__.py

@@ -7,6 +7,7 @@
 from .order_data import OrderResourceReader
 from .order_line_item_data import OrderLineItemResourceReader
 from .orderable_data import OrderableResourceReader
+from .price_changes_data import PriceChangesResourceReader
 from .program_data import ProgramResourceReader
 from .program_orderable_data import ProgramOrderableResourceReader
 from .proof_of_delivery_data import ProofOfDeliveryResourceReader

sigeca_data_export_microservice/app/domain/resources/price_changes_data.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+from uuid import uuid4
+
+from .abstract import ResourceReader
+from .util import schema_map, table_map, map_data
+from ...infrastructure import ChangeLogOperation
+
+
+class PriceChangesResourceReader(ResourceReader):
+    @classmethod
+    def read_table_name(cls):
+        return "price_changes"
+
+    @classmethod
+    def read_schema_name(cls):
+        return "referencedata"
+
+    def transform_data(self, df):
+        df = df.withColumn("occurreddate", df["occurreddate"].cast('string')) \
+            .withColumn("price", df["price"].cast("string"))
+
+        schema_name = self.read_schema_name()
+        table_name = self.read_table_name()
+
+        return df.rdd.map(self._to_payload(table_name, schema_name)).collect()
+
+    def _to_payload(self, table_name, schema_name):
+        def map_schema(row):
+            return {
+                "id": str(uuid4()),
+                "schema_name": schema_map[(schema_name, table_name)],
+                "table_name": table_map[(schema_name, table_name)],
+                "operation": ChangeLogOperation.SYNC.value,
+                "change_time": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
+                "row_data": map_data(row.asDict(), schema_name, table_name),
+            }
+
+        return map_schema

sigeca_data_export_microservice/app/domain/resources/util.py

@@ -6,6 +6,7 @@
     ("fulfillment", "orders"): "",
     ("fulfillment", "order_line_items"): "",
     ("referencedata", "orderables"): "",
+    ("referencedata", "price_changes"): "",
     ("referencedata", "programs"): "",
     ("referencedata", "program_orderables"): "",
     ("fulfillment", "proofs_of_delivery"): "",
@@ -29,6 +30,7 @@
     ("fulfillment", "orders"): "order",
     ("fulfillment", "order_line_items"): "order_line",
     ("referencedata", "orderables"): "product",
+    ("referencedata", "price_changes"): "price_changes",
     ("referencedata", "programs"): "program",
     ("referencedata", "program_orderables"): "program_product",
     ("fulfillment", "proofs_of_delivery"): "proof_of_delivery",
@@ -111,6 +113,13 @@
         "net_content": "netcontent",
         "round_to_zero": "roundtozero",
     },
+    ("referencedata", "price_changes"): {
+        "id": "id",
+        "program_product_id": "programorderableid",
+        "price": "price",
+        "author_id": "authorid",
+        "occurred_date": "occurreddate"
+    },
     ("referencedata", "programs"): {
         "id": "id",
         "code": "code",