同步代码

DjangoAppCenter/dtos/ret.py

@@ -0,0 +1,23 @@
+class BaseRet:
+    ret: int
+    message: str
+    error_code: str
+    data: dict
+
+    def to_dict(self):
+        return self.__dict__
+
+    @staticmethod
+    def success(message: str = None) -> dict:
+        base_ret = BaseRet()
+        base_ret.ret = 1
+        base_ret.message = message
+        return base_ret.to_dict()
+
+    @staticmethod
+    def fail(message: str = "System error", error_code: str = "10000") -> dict:
+        base_ret = BaseRet()
+        base_ret.ret = 0
+        base_ret.message = message
+        base_ret.error_code = error_code
+        return base_ret.to_dict()

DjangoAppCenter/middlewares/exception_catcher.py

@@ -0,0 +1,19 @@
+import logging
+
+from django.http import HttpResponseBadRequest, JsonResponse
+
+from DjangoAppCenter.dtos.ret import BaseRet
+from settings.profile import SettingsLoaderError
+
+logger = logging.getLogger("exception")
+
+
+class CustomException(object):
+    def process_exception(self, request, exception):
+        logger.exception(exception)
+
+        if isinstance(exception, SettingsLoaderError):
+            return JsonResponse(BaseRet.fail(exception.message))
+
+        else:
+            return HttpResponseBadRequest()

DjangoAppCenter/settings/__init__.py

@@ -157,4 +157,3 @@
         globals()[key].update(**value)
     else:
         globals().update(**{key: value})
-globals()

DjangoAppCenter/settings/profile.py

@@ -0,0 +1,58 @@
+import json
+import logging
+import os
+
+logger = logging.getLogger("admin")
+
+
+class SettingsLoaderError(Exception):
+    pass
+
+
+class SettingsLoader:
+
+    def __init__(self, path: str):
+        if not os.path.exists(path):
+            raise SettingsLoaderError("Profile path not found")
+
+        self.settings_path = path
+
+    def load(self):
+        try:
+            return json.loads(open(self.settings_path, encoding="utf-8").read())
+        except json.JSONDecodeError:
+            raise SettingsLoaderError("Json decode failed.") from json.JSONDecodeError
+
+    def merge(self, app: str, options: str):
+        if not options:
+            return
+        try:
+            options = json.loads(options)
+            if not isinstance(options, dict):
+                logger.error(f"App[{app}] 配置文件解析错误 {options}")
+                return
+
+            with open(CWD_SETTINGS_PATH, 'r', encoding="utf-8") as f:
+                profile = json.loads(f.read())
+                for key, value in options.items():
+                    if isinstance(value, (list, tuple)):
+                        try:
+                            profile.update(**{key: list(set(profile.get(key, []) + value))})
+                        except TypeError:
+                            profile.update(**{key: profile.get(key, []) + value})
+
+                    elif isinstance(value, dict):
+                        if not profile.get(key, None):
+                            profile[key] = {}
+                        profile[key].update(**value)
+
+                    else:
+                        profile.update(**{key: value})
+
+            with open(CWD_SETTINGS_PATH, 'w', encoding="utf-8") as f:
+                f.write(json.dumps(profile, ensure_ascii=False, indent=2))
+
+            logger.info(f"App[{app}] settings success injected")
+
+        except json.JSONDecodeError:
+            logger.error(f"App[{app}] 配置文件解析错误 {options}")

tests/test_loader.py

@@ -0,0 +1,10 @@
+import unittest
+
+from DjangoAppCenter.settings.loader import load_settings_from_file
+
+
+class Test(unittest.TestCase):
+
+    def test_load_settings_from_file(self):
+        settings = load_settings_from_file()
+