Merge pull request #617 from gisce/fix-4667-stop-multiprocess

Convert RES4667 to StopMultiProcess
gisce · Feb 21, 2024 · b229dce · b229dce
2 parents c3749c0 + 5b8b9a6
commit b229dce
Show file tree

Hide file tree

Showing 12 changed files with 111 additions and 81 deletions.
diff --git a/libcnmc/res_4667/CT.py b/libcnmc/res_4667/CT.py
@@ -8,10 +8,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f, get_name_ti, get_codigo_ccaa
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class CT(MultiprocessBased):
+class CT(StopMultiprocessBased):
     """
     Class to generate F6 of 4667
     """
@@ -57,6 +57,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 ct = O.GiscedataCnmcCt.read(item, fields_to_read)
@@ -76,10 +79,9 @@ def consumer(self):
                     ct["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/DES.py b/libcnmc/res_4667/DES.py
@@ -8,10 +8,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import get_name_ti, get_codigo_ccaa, format_f
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class DES(MultiprocessBased):
+class DES(StopMultiprocessBased):
     """
     Class to generate F6 of 4667
     """
@@ -57,6 +57,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 des = O.GiscedataCnmcDespatx.read(item, fields_to_read)
@@ -74,10 +77,9 @@ def consumer(self):
                     des["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/FIA.py b/libcnmc/res_4667/FIA.py
@@ -8,10 +8,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f, get_codigo_ccaa, get_name_ti
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class FIA(MultiprocessBased):
+class FIA(StopMultiprocessBased):
     """
     Class to generate F6 of 4667
     """
@@ -58,6 +58,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 fia = O.GiscedataCnmcFiabilitat.read(item, fields_to_read)
@@ -77,10 +80,9 @@ def consumer(self):
                     fia["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/LAT.py b/libcnmc/res_4667/LAT.py
@@ -8,10 +8,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import get_name_ti, get_codigo_ccaa, format_f
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class LAT(MultiprocessBased):
+class LAT(StopMultiprocessBased):
     """
     Class to generate F1 of 4667
     """
@@ -54,6 +54,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 linia = O.GiscedataCnmcLinies.read(item, fields_to_read)
@@ -86,10 +89,9 @@ def consumer(self):
                     linia["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/LBT.py b/libcnmc/res_4667/LBT.py
@@ -3,12 +3,12 @@
 
 import traceback
 
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import get_codigo_ccaa, get_name_ti, format_f
 
 
-class LBT(MultiprocessBased):
+class LBT(StopMultiprocessBased):
     """
     Class to generate F2 of 4667
     """
@@ -52,6 +52,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 lbt = O.GiscedataCnmcLiniesbt.read(item, fields_to_read)
@@ -74,10 +77,9 @@ def consumer(self):
                     lbt["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/MACRO.py b/libcnmc/res_4667/MACRO.py
@@ -4,11 +4,11 @@
 import traceback
 
 from libcnmc.res_4667.utils import get_resum_any_id
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 from libcnmc.utils import format_f
 
 
-class MACRO(MultiprocessBased):
+class MACRO(StopMultiprocessBased):
     """
     Class to generate F2 of 4667
     """
@@ -49,6 +49,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 macro = O.GiscedataCnmcResum_any.read(item, fields_to_read)
@@ -60,10 +63,9 @@ def consumer(self):
                     format_f(float(macro["macro_inc_demanda_sector"]), 2) or "0.00"
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/MAQ.py b/libcnmc/res_4667/MAQ.py
@@ -5,10 +5,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f, get_name_ti, get_codigo_ccaa
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class MAQ(MultiprocessBased):
+class MAQ(StopMultiprocessBased):
     """
     Class to generate F5 of 4667
     """
@@ -54,6 +54,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 maq = O.GiscedataCnmcMaquines.read(item, fields_to_read)
@@ -74,9 +77,9 @@ def consumer(self):
                     maq["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/Otros.py b/libcnmc/res_4667/Otros.py
@@ -5,10 +5,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import get_codigo_ccaa, format_f
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class Otros(MultiprocessBased):
+class Otros(StopMultiprocessBased):
     """
     Class to generate F3 of 4667
     """
@@ -51,6 +51,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 otro = O.GiscedataCnmcAltres.read(item, fields_to_read)
@@ -69,10 +72,9 @@ def consumer(self):
                     otro["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/POS.py b/libcnmc/res_4667/POS.py
@@ -6,9 +6,10 @@
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f, get_name_ti, get_codigo_ccaa
 from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class POS(MultiprocessBased):
+class POS(StopMultiprocessBased):
     """
     Class to generate F5 of 4667
     """
@@ -53,6 +54,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 pos = O.GiscedataCnmcPosicions.read(item, fields_to_read)
@@ -72,9 +76,9 @@ def consumer(self):
                     pos["actuacio_elegible_prtr"] or '',
                 ]
                 self.output_q.put(output)
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/PRO.py b/libcnmc/res_4667/PRO.py
@@ -5,10 +5,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f, get_codigo_ccaa
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class PRO(MultiprocessBased):
+class PRO(StopMultiprocessBased):
     """
     Class to generate proyectos of 4667
     """
@@ -87,6 +87,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
                 pro = O.GiscedataCnmcProjectes.read(item, fields_to_read)
                 codigo = pro["codi"]
@@ -110,11 +113,10 @@ def consumer(self):
                     pro["estado"]
                 ]
 
-
                 self.output_q.put(output)
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()
diff --git a/libcnmc/res_4667/RES.py b/libcnmc/res_4667/RES.py
@@ -5,10 +5,10 @@
 
 from libcnmc.res_4667.utils import get_resum_any_id
 from libcnmc.utils import format_f
-from libcnmc.core import MultiprocessBased
+from libcnmc.core import StopMultiprocessBased
 
 
-class RES(MultiprocessBased):
+class RES(StopMultiprocessBased):
     """
     Class to generate F2 of 4667
     """
@@ -70,6 +70,9 @@ def consumer(self):
         while True:
             try:
                 item = self.input_q.get()
+                if item == "STOP":
+                    self.input_q.task_done()
+                    break
                 self.progress_q.put(item)
 
                 resumen = O.GiscedataCnmcResum_any.read(item, fields_to_read)
@@ -100,10 +103,9 @@ def consumer(self):
                     resumen["num_proyectos_prtr"] or "0",
                 ]
                 self.output_q.put(output)
-
+                self.input_q.task_done()
             except Exception:
+                self.input_q.task_done()
                 traceback.print_exc()
                 if self.raven:
                     self.raven.captureException()
-            finally:
-                self.input_q.task_done()