src/risotto/dispatcher.py

@@ -260,67 +260,58 @@ class PublishDispatcher:
                                                             False,
                                                             False,
                                                             )
-        async with self.pool.acquire() as log_connection:
-            await log_connection.set_type_codec(
+        async with self.pool.acquire() as connection:
+            await connection.set_type_codec(
                 'json',
                 encoder=dumps,
                 decoder=loads,
                 schema='pg_catalog'
             )
-            async with log_connection.transaction():
-                risotto_context.log_connection = log_connection
-                async with self.pool.acquire() as connection:
-                    await connection.set_type_codec(
-                        'json',
-                        encoder=dumps,
-                        decoder=loads,
-                        schema='pg_catalog'
-                    )
-                    risotto_context.connection = connection
-                    for function_obj in self.messages[version][message]['functions']:
-                        function_name = function_obj['function'].__name__
-                        info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
+            risotto_context.connection = connection
+            for function_obj in self.messages[version][message]['functions']:
+                function_name = function_obj['function'].__name__
+                info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
+                try:
+                    async with connection.transaction():
                         try:
-                            async with connection.transaction():
-                                try:
-                                    await log.start(risotto_context,
-                                                    kwargs,
-                                                    info_msg,
-                                                    )
-                                    await self.check_message_type(risotto_context,
-                                                                  kwargs,
-                                                                  )
-                                    await self.launch(risotto_context,
-                                                      kwargs,
-                                                      config_arguments,
-                                                      function_obj,
-                                                      )
-                                    # log the success
-                                    await log.success(risotto_context)
-                                except CallError as err:
-                                    if get_config()['global']['debug']:
-                                        print_exc()
-                                    await log.failed(risotto_context,
-                                                     str(err),
-                                                     )
-                        except CallError:
-                            pass
-                        except Exception as err:
-                            # if there is a problem with arguments, log and do nothing
+                            await log.start(risotto_context,
+                                            kwargs,
+                                            info_msg,
+                                            )
+                            await self.check_message_type(risotto_context,
+                                                          kwargs,
+                                                          )
+                            await self.launch(risotto_context,
+                                              kwargs,
+                                              config_arguments,
+                                              function_obj,
+                                              )
+                            # log the success
+                            await log.success(risotto_context)
+                        except CallError as err:
                             if get_config()['global']['debug']:
                                 print_exc()
-                            async with self.pool.acquire() as connection:
-                                await connection.set_type_codec(
-                                    'json',
-                                    encoder=dumps,
-                                    decoder=loads,
-                                    schema='pg_catalog'
-                                )
-                                risotto_context.connection = connection
-                                async with connection.transaction():
-                                    await log.failed(risotto_context,
-                                                     str(err),
-                                                     )
+                            await log.failed(risotto_context,
+                                             str(err),
+                                             )
+                except CallError:
+                    pass
+                except Exception as err:
+                    # if there is a problem with arguments, log and do nothing
+                    if get_config()['global']['debug']:
+                        print_exc()
+                    async with self.pool.acquire() as connection:
+                        await connection.set_type_codec(
+                            'json',
+                            encoder=dumps,
+                            decoder=loads,
+                            schema='pg_catalog'
+                        )
+                        risotto_context.connection = connection
+                        async with connection.transaction():
+                            await log.failed(risotto_context,
+                                             str(err),
+                                             )
 
 
 class Dispatcher(register.RegisterDispatcher,