Elastic Bulk Document Creation

tdrs-backend/docker-compose.yml

@@ -45,12 +45,25 @@ services:
       # Copy in the Localstack setup script to configure any buckets needed
       - ../scripts/localstack-setup.sh:/docker-entrypoint-initaws.d/localstack-setup.sh
 
+  kibana:
+    image: elastic/kibana:7.17.10
+    ports:
+      - 5601:5601
+    environment:
+      - xpack.security.encryptionKey="something_at_least_32_characters"
+      - xpack.security.session.idleTimeout="1h"
+      - xpack.security.session.lifespan="30d"
+    volumes:
+      - ./kibana.yml:/usr/share/kibana/config/kibana.yml
+    depends_on:
+      - elastic
+
   elastic:
     image: elasticsearch:7.17.6
     environment:
       - discovery.type=single-node
-      - xpack.security.enabled=false
       - logger.discovery.level=debug
+      - xpack.security.enabled=false
     ports:
       - 9200:9200
       - 9300:9300

tdrs-backend/kibana.yml

@@ -0,0 +1,2 @@
+elasticsearch.hosts: ["http://elastic:9200"]
+server.host: kibana

tdrs-backend/tdpservice/parsers/parse.py

@@ -55,22 +55,29 @@ def parse_datafile(datafile):
 
     return errors
 
-
 def bulk_create_records(unsaved_records, line_number, header_count, batch_size=10000, flush=False):
     """Bulk create passed in records."""
     if (line_number % batch_size == 0 and header_count > 0) or flush:
         logger.debug("Bulk creating records.")
         try:
-            num_created = 0
-            num_expected = 0
-            for model, records in unsaved_records.items():
-                num_expected += len(records)
-                num_created += len(model.objects.bulk_create(records))
-            if num_created != num_expected:
-                logger.error(f"Bulk create only created {num_created}/{num_expected}!")
+            num_db_records_created = 0
+            num_expected_db_records = 0
+            num_elastic_records_created = 0
+            for document, records in unsaved_records.items():
+                num_expected_db_records += len(records)
+                created_objs = document.Django.model.objects.bulk_create(records)
+                num_elastic_records_created += document.update(created_objs)[0]
+                num_db_records_created += len(created_objs)
+            if num_db_records_created != num_expected_db_records:
+                logger.error(f"Bulk Django record creation only created {num_db_records_created}/" +
+                             f"{num_expected_db_records}!")
+            elif num_elastic_records_created != num_expected_db_records:
+                logger.error(f"Bulk Elastic document creation only created {num_elastic_records_created}/" +
+                             f"{num_expected_db_records}!")
             else:
-                logger.info(f"Created {num_created}/{num_expected} records.")
-            return num_created == num_expected, {}
+                logger.info(f"Created {num_db_records_created}/{num_expected_db_records} records.")
+            return num_db_records_created == num_expected_db_records and \
+                num_elastic_records_created == num_expected_db_records, {}
         except DatabaseError as e:
             logger.error(f"Encountered error while creating datafile records: {e}")
             return False, unsaved_records
@@ -107,7 +114,8 @@ def evaluate_trailer(datafile, trailer_count, multiple_trailer_errors, is_last_l
 def rollback_records(unsaved_records, datafile):
     """Delete created records in the event of a failure."""
     logger.info("Rolling back created records.")
-    for model in unsaved_records:
+    for document in unsaved_records:
+        model = document.Django.model
         num_deleted, models = model.objects.filter(datafile=datafile).delete()
         logger.debug(f"Deleted {num_deleted} records of type: {model}.")
 
@@ -198,7 +206,7 @@ def parse_datafile_lines(datafile, program_type, section, is_encrypted):
             if record:
                 s = schema_manager.schemas[i]
                 record.datafile = datafile
-                unsaved_records.setdefault(s.model, []).append(record)
+                unsaved_records.setdefault(s.document, []).append(record)
 
         all_created, unsaved_records = bulk_create_records(unsaved_records, line_number, header_count,)
         unsaved_parser_errors, num_errors = bulk_create_errors(unsaved_parser_errors, num_errors)

tdrs-backend/tdpservice/parsers/row_schema.py

@@ -12,13 +12,13 @@ class RowSchema:
 
     def __init__(
             self,
-            model=dict,
+            document,
             preparsing_validators=[],
             postparsing_validators=[],
             fields=[],
-            quiet_preparser_errors=False
+            quiet_preparser_errors=False,
             ):
-        self.model = model
+        self.document = document
         self.preparsing_validators = preparsing_validators
         self.postparsing_validators = postparsing_validators
         self.fields = fields
@@ -90,7 +90,7 @@ def run_preparsing_validators(self, line, generate_error):
 
     def parse_line(self, line):
         """Create a model for the line based on the schema."""
-        record = self.model()
+        record = self.document.Django.model() if self.document is not None else dict()
 
         for field in self.fields:
             value = field.parse_value(line)

tdrs-backend/tdpservice/parsers/schema_defs/header.py

@@ -7,7 +7,7 @@
 
 
 header = RowSchema(
-    model=dict,
+    document=None,
     preparsing_validators=[
         validators.hasLength(
             23,

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m1.py

@@ -5,12 +5,12 @@
 from tdpservice.parsers.fields import Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.ssp import SSP_M1
+from tdpservice.search_indexes.documents.ssp import SSP_M1DataSubmissionDocument
 
 m1 = SchemaManager(
     schemas=[
         RowSchema(
-            model=SSP_M1,
+            document=SSP_M1DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(150),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m2.py

@@ -6,13 +6,13 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.ssp import SSP_M2
+from tdpservice.search_indexes.documents.ssp import SSP_M2DataSubmissionDocument
 
 
 m2 = SchemaManager(
     schemas=[
         RowSchema(
-            model=SSP_M2,
+            document=SSP_M2DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(150),
             ],
@@ -78,7 +78,7 @@
                     result_field='EDUCATION_LEVEL',
                     result_function=validators.or_validators(
                         validators.isInStringRange(1, 16),
-                        validators.isInStringRange(98, 99)
+                        validators.isInStringRange(98, 99),
                     ),
                 ),
                 validators.if_then_validator(
@@ -367,8 +367,7 @@
                     required=False,
                     validators=[
                         validators.or_validators(
-                            validators.isInLimits(0, 16),
-                            validators.isInLimits(98, 99)
+                            validators.isInLimits(0, 16), validators.isInLimits(98, 99)
                         )
                     ]
                 ),
@@ -414,7 +413,7 @@
                         validators.or_validators(
                             validators.isInLimits(1, 4),
                             validators.isInLimits(6, 9),
-                            validators.isInLimits(11, 12)
+                            validators.isInLimits(11, 12),
                         )
                     ]
                 ),

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m3.py

@@ -6,10 +6,10 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.ssp import SSP_M3
+from tdpservice.search_indexes.documents.ssp import SSP_M3DataSubmissionDocument
 
 first_part_schema = RowSchema(
-    model=SSP_M3,
+    document=SSP_M3DataSubmissionDocument(),
     preparsing_validators=[
         validators.notEmpty(start=19, end=60),
     ],
@@ -315,7 +315,7 @@
 )
 
 second_part_schema = RowSchema(
-    model=SSP_M3,
+    document=SSP_M3DataSubmissionDocument(),
     quiet_preparser_errors=True,
     preparsing_validators=[
         validators.notEmpty(start=60, end=101),

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m4.py

@@ -5,12 +5,12 @@
 from tdpservice.parsers.fields import Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.ssp import SSP_M4
+from tdpservice.search_indexes.documents.ssp import SSP_M4DataSubmissionDocument
 
 m4 = SchemaManager(
     schemas=[
         RowSchema(
-            model=SSP_M4,
+            document=SSP_M4DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(66),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m5.py

@@ -6,13 +6,13 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.ssp import SSP_M5
+from tdpservice.search_indexes.documents.ssp import SSP_M5DataSubmissionDocument
 
 
 m5 = SchemaManager(
     schemas=[
         RowSchema(
-            model=SSP_M5,
+            document=SSP_M5DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(66),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m6.py

@@ -6,10 +6,10 @@
 from ...fields import Field, TransformField
 from ...row_schema import RowSchema
 from ... import validators
-from tdpservice.search_indexes.models.ssp import SSP_M6
+from tdpservice.search_indexes.documents.ssp import SSP_M6DataSubmissionDocument
 
 s1 = RowSchema(
-    model=SSP_M6,
+    document=SSP_M6DataSubmissionDocument(),
     preparsing_validators=[
         validators.hasLength(259),
     ],
@@ -170,7 +170,7 @@
 )
 
 s2 = RowSchema(
-    model=SSP_M6,
+    document=SSP_M6DataSubmissionDocument(),
     preparsing_validators=[
         validators.hasLength(259),
     ],
@@ -331,7 +331,7 @@
 )
 
 s3 = RowSchema(
-    model=SSP_M6,
+    document=SSP_M6DataSubmissionDocument(),
     preparsing_validators=[
         validators.hasLength(259),
     ],
@@ -492,10 +492,4 @@
 )
 
 
-m6 = SchemaManager(
-    schemas=[
-        s1,
-        s2,
-        s3
-    ]
-)
+m6 = SchemaManager(schemas=[s1, s2, s3])

tdrs-backend/tdpservice/parsers/schema_defs/ssp/m7.py

@@ -5,7 +5,7 @@
 from ...row_schema import RowSchema
 from ...transforms import calendar_quarter_to_rpt_month_year
 from ... import validators
-from tdpservice.search_indexes.models.ssp import SSP_M7
+from tdpservice.search_indexes.documents.ssp import SSP_M7DataSubmissionDocument
 
 schemas = []
 
@@ -20,7 +20,7 @@
 for i in range(1, 31):
     schemas.append(
         RowSchema(
-            model=SSP_M7,
+            document=SSP_M7DataSubmissionDocument(),
             quiet_preparser_errors=i > 1,
             preparsing_validators=[
                 validators.hasLength(247),

tdrs-backend/tdpservice/parsers/schema_defs/tanf/t1.py

@@ -4,13 +4,13 @@
 from tdpservice.parsers.fields import Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.tanf import TANF_T1
+from tdpservice.search_indexes.documents.tanf import TANF_T1DataSubmissionDocument
 
 
 t1 = SchemaManager(
     schemas=[
         RowSchema(
-            model=TANF_T1,
+            document=TANF_T1DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(156),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/tanf/t2.py

@@ -6,13 +6,13 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.tanf import TANF_T2
+from tdpservice.search_indexes.documents.tanf import TANF_T2DataSubmissionDocument
 
 
 t2 = SchemaManager(
     schemas=[
         RowSchema(
-            model=TANF_T2,
+            document=TANF_T2DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(156),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/tanf/t3.py

@@ -6,11 +6,11 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.tanf import TANF_T3
+from tdpservice.search_indexes.documents.tanf import TANF_T3DataSubmissionDocument
 
 
 child_one = RowSchema(
-    model=TANF_T3,
+    document=TANF_T3DataSubmissionDocument(),
     preparsing_validators=[
         validators.notEmpty(start=19, end=60),
     ],
@@ -313,7 +313,7 @@
 )
 
 child_two = RowSchema(
-    model=TANF_T3,
+    document=TANF_T3DataSubmissionDocument(),
     quiet_preparser_errors=True,
     preparsing_validators=[
         validators.notEmpty(start=60, end=101),

tdrs-backend/tdpservice/parsers/schema_defs/tanf/t4.py

@@ -5,13 +5,13 @@
 from tdpservice.parsers.fields import Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.tanf import TANF_T4
+from tdpservice.search_indexes.documents.tanf import TANF_T4DataSubmissionDocument
 
 
 t4 = SchemaManager(
     schemas=[
         RowSchema(
-            model=TANF_T4,
+            document=TANF_T4DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(71),
             ],

tdrs-backend/tdpservice/parsers/schema_defs/tanf/t5.py

@@ -6,13 +6,13 @@
 from tdpservice.parsers.fields import TransformField, Field
 from tdpservice.parsers.row_schema import RowSchema
 from tdpservice.parsers import validators
-from tdpservice.search_indexes.models.tanf import TANF_T5
+from tdpservice.search_indexes.documents.tanf import TANF_T5DataSubmissionDocument
 
 
 t5 = SchemaManager(
     schemas=[
         RowSchema(
-            model=TANF_T5,
+            document=TANF_T5DataSubmissionDocument(),
             preparsing_validators=[
                 validators.hasLength(71),
             ],