Update channel message.

src/snek/service/channel_message.py

@@ -2,7 +2,18 @@ from snek.system.service import BaseService
 from snek.system.template import sanitize_html
 import time
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ProcessPoolExecutor
+import json
+
+from jinja2 import Environment, FileSystemLoader
+global jinja2_env
+import pathlib 
+template_path = pathlib.Path(__file__).parent.parent.joinpath("templates")
+
+
+def render(context):
+    template =jinja2_env.get_template("message.html")
+    return sanitize_html(template.render(**context))
 
 
 
@@ -13,11 +24,18 @@ class ChannelMessageService(BaseService):
         super().__init__(*args, **kwargs)
         self._configured_indexes = False
         self._executor_pools = {}
+        global jinja2_env
+        jinja2_env = self.app.jinja2_env
+        self._max_workers = 1
+    def get_or_create_executor(self, uid):
+        if not uid in self._executor_pools:
+            self._executor_pools[uid] = ProcessPoolExecutor(max_workers=5)
+        return self._executor_pools[uid]
     
-    def get_or_create_executor(self, user_uid):
+    def delete_executor(self, uid):
-        if not user_uid in self._executor_pools:
+        if uid in self._executor_pools:
-            self._executor_pools[user_uid] = ThreadPoolExecutor(max_workers=1)
+            self._executor_pools[uid].shutdown()
-        return self._executor_pools[user_uid]
+            del self._executor_pools[uid]
 
     async def maintenance(self):
         args = {}
@@ -81,9 +99,12 @@ class ChannelMessageService(BaseService):
         )
         loop = asyncio.get_event_loop()
         try:
-            template = self.app.jinja2_env.get_template("message.html")
+
-            model["html"] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), lambda: template.render(**context))
+            context = json.loads(json.dumps(context, default=str)) 
-            model['html'] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), lambda:sanitize_html(model['html']))
+
+
+            model["html"] = await loop.run_in_executor(self.get_or_create_executor(model["uid"]), render,context)
+            #model['html'] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), sanitize_html,model['html'])
         except Exception as ex:
             print(ex, flush=True)
 
@@ -102,6 +123,8 @@ class ChannelMessageService(BaseService):
                         ["deleted_at"], unique=False
                     )
             self._configured_indexes = True
+            if model['is_final']:
+                self.delete_executor(model['uid'])
             return model
         raise Exception(f"Failed to create channel message: {model.errors}.")
 
@@ -138,12 +161,15 @@ class ChannelMessageService(BaseService):
                 "color": user["color"],
             }
         )
-        template = self.app.jinja2_env.get_template("message.html")
+        context = json.loads(json.dumps(context, default=str)) 
-        
         loop = asyncio.get_event_loop()
-        model["html"] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), lambda: template.render(**context))
+        model["html"] = await loop.run_in_executor(self.get_or_create_executor(model["uid"]), render, context)
-        model['html'] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), lambda: sanitize_html(model['html']))
+        #model['html'] = await loop.run_in_executor(self.get_or_create_executor(user["uid"]), sanitize_html,model['html'])
-        return await super().save(model)
+
+        result = await super().save(model)
+        if model['is_final']:
+            self.delete_executor(model['uid'])
+        return result 
 
     async def offset(self, channel_uid, page=0, timestamp=None, page_size=30):
         channel = await self.services.channel.get(uid=channel_uid)