feat: add settings module

.gitignore

@@ -0,0 +1,5 @@
+__pycache__
+.idea
+logs
+.venv
+.vscode

Makefile

@@ -0,0 +1,2 @@
+test:
+	python -m unitest test/settings/test_settings.py

appmanage/api/exception/command_exception.py

@@ -3,3 +3,8 @@ class CommandException(Exception):
         self.code = code
         self.message = message
         self.detail = detail
+
+
+class MissingConfigException(CommandException):
+
+    pass

appmanage/api/settings/settings.py

@@ -0,0 +1,50 @@
+from api.utils.helper import Singleton
+
+
+__all__ = ['settings']
+
+
+class Settings(object):
+
+    __metaclass__ = Singleton
+
+    def __init__(self):
+        self._config = {}
+        self.config_file = '/usr/src/app/config/settings.conf'
+
+    def init_config_from_file(self, config_file: str=None):
+        if config_file:
+            self.config_file = config_file
+        try:
+            with open(config_file, 'r') as f:
+                data = f.readlines()
+        except Exception:
+            data = []
+        for i in data:
+            i = i.replace('\n', '').replace('\r\n', '')
+            key, value = i.split('=')
+            if self._config.get(key) != value:
+                self._config[key] = value
+
+    def update_setting(self, key: str, value: str):
+        self._config[key] = value
+        self.flush_config()
+
+    def get_setting(self, key: str, default=None):
+        return self._config.get(key, default)
+
+    def list_all_settings(self) -> list:
+        return self._config
+
+    def delete_setting(self, key: str, value: str):
+        if key in self._config:
+            del self._config[key]
+
+    def flush_config(self):
+        with open(self.config_file, 'w') as f:
+            for key, value in self._config.items():
+                f.write(f'{key}={value}\n')
+
+
+
+settings = Settings()

appmanage/api/utils/const.py

@@ -9,6 +9,7 @@ ERROR_CONFIG_NGINX="Nginx.Configure.Error"
 ERROR_SERVER_COMMAND = "Server.Container.Error"
 ERROR_SERVER_SYSTEM = "Server.SystemError"
 ERROR_SERVER_RESOURCE = "Server.ResourceError"
+ERROR_SERVER_CONFIG_MISSING = "Server.Config.NotFound"
 
 # 错误信息定义
 ERRORMESSAGE_CLIENT_PARAM_BLANK = "Client.Parameter.Blank.Error"

appmanage/api/utils/helper.py

@@ -0,0 +1,6 @@
+class Singleton(type):
+    _instances = {}
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]

appmanage/api/v1/routers/apps.py

@@ -11,6 +11,7 @@ from api.service import manage, db
 from api.utils import shell_execute, const
 from api.utils.common_log import myLogger
 from api.exception.command_exception import CommandException
+from api.settings.settings import settings
 
 router = APIRouter()
 
@@ -76,6 +77,13 @@ rd_appstore = rd_s + appstore_update + rd_e
 rd_auto_list = rd_s + auto + rd_e
 rd_user_list = rd_s + user + rd_e
 rd_updateuser_list=rd_s + updateuser + rd_e
+
+
+class SettingItem(BaseModel):
+
+    key: str = Field(description="配置项")
+    value: str = Field(description="配置项的取值")
+
 @router.api_route("/AppStatus", methods=["GET", "POST"], summary="获取指定APP的信息",
                   response_description=rd_status,
                   response_model=Response)
@@ -526,6 +534,27 @@ def AppUpdateUser(request: Request,user_name: Optional[str] = Query(default=None
 
     return response
 
+
+@router.api_route("/AppListSettings", methods=['GET', 'POST'], summary="获取配置信息")
+def list_settings():
+    items = settings.list_all_settings()
+    return [{
+        'key': key,
+        'value': value
+    } for key, value in items.items()]
+
+
+@router.api_route("/AppUpdateSettings", methods=['GET', 'POST'], summary="创建或者更新配置信息")
+def create_or_update_settings(item: SettingItem):
+    settings.update_setting(item.key, item.value)
+
+
+
+@router.api_route("/AppDeleteSettings", methods=['GET', 'POST'], summary="删除配置信息")
+def delete_settings(item: SettingItem):
+    settings.delete_setting(item.key, item.value)
+
+
 def get_headers(request):
     headers = request.headers
     try:

appmanage/main.py

@@ -1,7 +1,6 @@
-import api.v1.api as api_router_v1
+import argparse
 import uvicorn
-from api.utils.common_log import myLogger
-from api.utils import shell_execute
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
@@ -11,10 +10,19 @@ from fastapi.openapi.docs import (
     get_swagger_ui_oauth2_redirect_html,
 )
 
+import api.v1.api as api_router_v1
+
+from api.utils.common_log import myLogger
+from api.utils import shell_execute
+from api.settings.settings import settings
+
+
 myLogger.info_logger("Start server...")
 app = FastAPI(docs_url=None, redoc_url=None, openapi_url="/")
 
+
 def get_app():
+    settings.init_config_from_file()
     origins = [
         "http://localhost",
         "http://localhost:9090",
@@ -45,6 +53,7 @@ async def custom_swagger_ui_html():
 async def swagger_ui_redirect():
     return get_swagger_ui_oauth2_redirect_html()
 
+
 @app.get("/redoc", include_in_schema=False)
 async def redoc_html():
     return get_redoc_html(
@@ -54,4 +63,9 @@ async def redoc_html():
     )
 
 if __name__ == "__main__":
-    uvicorn.run("main:get_app", host='0.0.0.0', port=5000, reload=True)
+    parser = argparse.ArgumentParser(description='websoft9')
+    parser.add_argument("--port", type=int, dest='port', default=5000, metavar="port")
+    parser.add_argument("--config", type=str, dest="config_file", required=True)
+    args = parser.parse_args()
+    settings.init_config_from_file(config_file=args.config_file)
+    uvicorn.run("main:get_app", host='0.0.0.0', port=args.port, reload=True)

test/settings/test_settings.py

@@ -0,0 +1,46 @@
+import os
+import tempfile
+import unittest
+
+from appmanage.api.settings.settings import settings
+
+
+class TestSettings(unittest.TestCase):
+    fd = None
+
+    @classmethod
+    def setUpClass(cls):
+        fd = tempfile.NamedTemporaryFile("w")
+        print(fd.name)
+        fd.write("a=b\nc=d\n")
+        fd.flush()
+        settings.init_config_from_file(fd.name)
+
+    def test_get_config(self):
+        self.assertEqual(settings.get_setting("a"), "b")
+        self.assertTrue(settings.get_setting("e") is None)
+
+    def test_update_config(self):
+        self.assertEqual(settings.get_setting("a"), "b")
+        settings.update_setting("a", "i")
+        self.assertEqual(settings.get_setting("a"), "i")
+
+    def test_list_settings(self):
+        data = settings.list_all_settings()
+        self.assertTrue(data is not None)
+
+    def test_delete_config(self):
+        settings.update_setting("x", "y")
+        v = settings.get_setting("x")
+        self.assertTrue(v is not None)
+        settings.delete_setting("x", v)
+        self.assertTrue(settings.get_setting("x") is None)
+        
+    @classmethod
+    def tearDownClass(cls):
+        if cls.fd:
+            cls.fd.close()
+
+
+if __name__ == '__main__':
+    unittest.main()