v0.0.60 clear_by_source to happen before data processing

Author: akariv

Dockerfile

@@ -10,7 +10,7 @@ RUN mkdir -p /var/dgp
 
 FROM build
 
-RUN pip install --no-cache-dir dgp_server==0.0.59
+RUN pip install --no-cache-dir dgp_server==0.0.60
 
 ENV SERVER_MODULE=dgp_server.server:app
 

dgp_server/VERSION

@@ -1 +1 @@
-0.0.59
+0.0.60

dgp_server/publish_flow.py

@@ -13,28 +13,27 @@
 
 def clear_by_source(engine: Engine, table_name, source):
     index_name = table_name + '__s'
+
     def func(package):
+        with engine.connect() as conn:
+            s = text('create index concurrently ' +
+                        f'{index_name} on {table_name} (_source)')
+            try:
+                logger.info('CREATING INDEX')
+                conn.execute(s)
+                logger.info('DONE CREATING INDEX')
+            except ProgrammingError as e:
+                logger.error('Failed to create index %s', e)
+            s = text(f'delete from {table_name} where _source=:source'
+                        ).params(source=source)
+            try:
+                logger.info('DELETING PAST ROWS')
+                conn.execute(s)
+                logger.info('DONE DELETING')
+            except ProgrammingError as e:
+                logger.error('Failed to remove rows %s', e)
         yield package.pkg
-        for i, resource in enumerate(package):
-            if i == 0:
-                with engine.connect() as conn:
-                    s = text('create index concurrently ' +
-                             f'{index_name} on {table_name} (_source)')
-                    try:
-                        logger.info('CREATING INDEX')
-                        conn.execute(s)
-                        logger.info('DONE CREATING INDEX')
-                    except ProgrammingError as e:
-                        logger.error('Failed to create index %s', e)
-                    s = text(f'delete from {table_name} where _source=:source'
-                             ).params(source=source)
-                    try:
-                        logger.info('DELETING PAST ROWS')
-                        conn.execute(s)
-                        logger.info('DONE DELETING')
-                    except ProgrammingError as e:
-                        logger.error('Failed to remove rows %s', e)
-            yield resource
+        yield from package
 
     return func