Ingest donation and shifts restore

API for 360 returns donations and shifts

Author: urirot

src/server/api/common_api.py

@@ -37,24 +37,21 @@ def get_360(matching_id):
         query_result = connection.execute(query, matching_id=matching_id)
 
         result["contact_details"] = [dict(row) for row in query_result]
-        result["shifts"] = []
-        result["donations"] = []
-        result["adoptions"] = []
 
-        # todo: complete retrieving details for response
-        for row in query_result:
+        for row in result["contact_details"]:
+            if row["source_type"] == "salesforcecontacts":
+                donations_query = text("select * from salesforcedonations where contact_id like :salesforcecontacts_id")
+                query_result = connection.execute(donations_query, salesforcecontacts_id=row["source_id"] + "%")
+                salesforcedonations_results = [dict(row) for row in query_result]
+                result['donations'] = salesforcedonations_results
+
             if row["source_type"] == "volgistics":
-                query = text("select * from volgisticsshifts where number = :volgistics_id")
-                query_result = connection.execute(query, volgistics_id=row["source_id"])
-                result["shifts"] += [dict(row) for row in query_result]
-
-        '''
-        query = text("select * from salesforcedonations where contact_id = :salesforcecontacts_id")
-        query_result = connection.execute(query, salesforcecontacts_id=salesforcecontacts_id)
-        salesforcedonations_results = [dict(row) for row in query_result]
-
-        if salesforcedonations_results:
-            result['salesforcedonations'] = salesforcedonations_results
-        '''
+                shifts_query = text("select * from volgisticsshifts where number = :volgistics_id")
+                query_result = connection.execute(shifts_query, volgistics_id=row["source_id"])
+                volgisticsshifts_results = [dict(row) for row in query_result]
+                result['shifts'] = volgisticsshifts_results
+
+            #todo: add adoptions
+
 
         return jsonify({'result': result})

src/server/datasource_manager.py

@@ -94,16 +94,16 @@ def volgistics_address(index, street):
         "state": "mailing_state_province",
         "zip": "mailing_zip_postal_code",
         "others": {
-            "additional_sources": [{
-                "salesforcedonations": {
-                        'should_drop_first_column': True
-                    }
-                }
-            ],
             "should_drop_first_column": True
         }
 
     },
+    "salesforcedonations": {
+        "parent": "salesforcecontacts",
+        "others": {
+            "should_drop_first_column": True
+        }
+    },
     "shelterluvpeople": {
         "source_id": "id",
         "first_name": "firstname",
@@ -131,19 +131,13 @@ def volgistics_address(index, street):
         "state": "state",
         "zip": "zip",
         "others": {
-            "additional_sources": [{
-                    "volgisticsshifts": {
-                        'should_drop_first_column': True
-                    }
-                }
-            ],
             "should_drop_first_column": True
         }
-
+    },
+    "volgisticsshifts": {
+        "parent": "volgistics",
+        "others": {
+            "should_drop_first_column": True
+        }
     }
 }
-
-
-
-
-

src/server/models.py

@@ -1,6 +1,6 @@
 import datetime
 
-from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
+from sqlalchemy import Column, Integer, String, DateTime
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.declarative import declarative_base
 
@@ -89,42 +89,4 @@ class Volgistics(Base):
     json = Column(JSONB)
 
 
-class SalesForceDonations(Base):
-    __tablename__ = "salesforcedonations"
 
-    _id = Column(Integer, primary_key=True)
-    recurring_donor = Column(String)
-    opportunity_owner = Column(String)
-    account_id = Column(String)
-    account_name = Column(String)
-    opportunity_id = Column(String)
-    opportunity_name = Column(String)
-    stage = Column(String)
-    fiscal_period = Column(String)
-    amount = Column(String)
-    probability = Column(String)
-    age = Column(String)
-    close_date = Column(String)
-    created_date = Column(String)
-    next_step = Column(String)
-    lead_source = Column(String)
-    type = Column(String)
-    source = Column(String)
-    contact_id = Column(String)
-    primary_campaign_source = Column(String)
-
-
-class Volgistics_Shifts(Base):
-    __tablename__ = 'volgisticsshifts'
-
-    _id = Column(Integer, primary_key=True)
-    number = Column(String)
-    site = Column(String)
-    place = Column(String)
-    assignment = Column(String)
-    role = Column(String)
-    from_date = Column('from', DateTime)
-    to = Column(DateTime)
-    spare_date = Column(String)
-    spare_chechbox = Column(String)
-    coordinator = Column(String)

src/server/pipeline/clean_and_load_data.py

@@ -9,7 +9,7 @@
 from config import CURRENT_SOURCE_FILES_PATH
 
 
-def start(pdp_contacts_df, file_path_list):
+def start(connection, pdp_contacts_df, file_path_list):
     result = pd.DataFrame(columns=pdp_contacts_df.columns)
 
     for uploaded_file in file_path_list:
@@ -26,12 +26,16 @@ def start(pdp_contacts_df, file_path_list):
         normalization_without_others = copy.deepcopy(SOURCE_NORMALIZATION_MAPPING[table_name])
         normalization_without_others.pop("others")  # copy avoids modifying the imported mapping
 
-        source_df = create_normalized_df(df, normalization_without_others, table_name)
+        if "parent" not in normalization_without_others:
+            source_df = create_normalized_df(df, normalization_without_others, table_name)
+
+            if result.empty:
+                result = source_df
+            else:
+                result = pd.concat([result, source_df])
 
-        if result.empty:
-            result = source_df
         else:
-            result = pd.concat([result, source_df])
+            df.to_sql(table_name, connection, index=False, if_exists='append')
 
         current_app.logger.info('   - Finish load_paws_data on: ' + uploaded_file)
 

src/server/pipeline/flow_script.py

@@ -23,7 +23,7 @@ def start_flow():
             # Clean the input data and normalize
             # input - existing files in path
             # output - normalized object of all entries
-            normalized_data = clean_and_load_data.start(pdp_contacts_df, file_path_list)
+            normalized_data = clean_and_load_data.start(connection, pdp_contacts_df, file_path_list)
 
             # Standardize column data types
             # If additional inconsistencies are encountered, may need to enforce the schema of