app manage

api/main.py

@@ -1,15 +0,0 @@
-from typing import Optional
-
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/")
-def read_root():
-    return {"Hello": "World"}
-
-
-@app.get("/items/{item_id}")
-def read_item(item_id: int, q: Optional[str] = None):
-    return {"item_id": item_id, "q": q}

api/requirements.txt

@@ -1,2 +0,0 @@
-fastapi
-uvicorn

appmanage/Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10-slim
+LABEL maintainer="Websoft9<help@websoft9.com>"
+
+# Create api directory
+WORKDIR /usr/src/app
+
+# Copy sw and install pip dependencies
+COPY api ./api
+COPY data/public_schools.db ./data/public_schools.db
+COPY requirements.txt main.py docker-entrypoint.sh ./
+RUN pip install -r requirements.txt
+
+# Starts our application
+CMD ["uvicorn", "main:get_app", "--host", "0.0.0.0", "--port", "5000", "--log-level", "info"]
+
+# Expose the port in which the application will be deployed
+EXPOSE 5000

appmanage/api/model/app.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class App(BaseModel):
+    id: int
+    name: str
+    status_code: int
+    status: str
+    port: int
+    volume: str
+    url: str

appmanage/api/service/manage.py

@@ -0,0 +1,50 @@
+import os, io, sys, platform, shutil, time, subprocess, json, datetime
+
+from api.utils import shell_execute
+
+# 获取我的信息的所有信息
+def get_my_app():
+
+    my_cmd = my_app()
+
+    apps_info = shell_execute.execute_CommandReturn(my_cmd)
+
+    return apps_info
+
+# 生成创建 App 的命令
+def my_app():
+
+    my_app_cmd = ''
+    my_app_cmd = "sudo su && docker compose ls"
+
+    return my_app_cmd
+
+
+# 生成创建 App 的命令
+def create_app(app_name):
+    print(app_name)
+
+    create_cmd = ''
+    create_cmd = "sudo su && cd /data/apps/" + app_name + " && docker compose up -d"
+
+    return create_cmd
+
+
+# 生成启动 App 的命令
+def start_app(app_name):
+    print(app_name)
+
+    start_cmd = ''
+    start_cmd = "sudo su && docker compose start" + app_name
+
+    return start_cmd
+
+
+# 生成停止 App 的命令
+def stop_app(app_name):
+    print(app_name)
+
+    stop_cmd = ''
+    stop_cmd = "sudo su && docker compose stop" + app_name
+
+    return stop_cmd

appmanage/api/utils/shell_execute.py

@@ -0,0 +1,47 @@
+import os, io, sys, platform, shutil, time, subprocess, json, datetime
+
+# 执行Shell命令，处理报错和超时，并有返回值
+def execute_Command(cmd_str, timeout=60, timeinner=3, retry=True):
+    print(cmd_str)
+    time_out = 0
+    status = False
+
+    while time_out < timeout:
+        out_str = subprocess.getstatusoutput(cmd_str)
+        print(out_str)
+        if out_str[0] == 0 and out_str[1].find('ERROR') == -1 and out_str[1].find('error') == -1:
+            status = True
+            print('\nExecute successfully')
+            break
+        elif retry:
+            print('\nTry again')
+            time.sleep(timeinner)
+            time_out = time_out + timeinner
+        else:
+            time_out = timeout
+
+    if not status:
+        print('\n此次任务执行有异常，请仔细排查')
+
+# 执行Shell命令，处理报错和超时，并有返回值
+def execute_CommandReturn(cmd_str, timeout=30, timeinner=3):
+    print(cmd_str)
+    time_out = 0
+
+    while time_out < timeout:
+        out_str = subprocess.getstatusoutput(cmd_str)
+        print(out_str)
+        if out_str[0] == 0 and out_str[1].find('ERROR') == -1 and out_str[1].find('error') == -1:
+            # 去掉\n和"
+            # 返回值是元组，不能修改，需要赋值给变量
+            temp_str = out_str[1]
+            temp_str = temp_str.strip('\n')
+            temp_str = temp_str.strip('"')
+            time_out = timeout
+            return temp_str
+        else:
+            time.sleep(timeinner)
+            time_out = time_out + timeinner
+    print('\n此次任务执行失败，请根据下面错误原因排查：')
+    print(out_str)
+

appmanage/api/v1/api.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from api.v1.routers import health,apps
+
+
+def get_api():
+    api_router = APIRouter()
+    api_router.include_router(health.router)
+    api_router.include_router(apps.router, prefix="/apps")
+    return api_router

appmanage/api/v1/routers/apps.py

@@ -0,0 +1,24 @@
+from typing import Optional, List
+
+from fastapi import APIRouter, status, Depends
+from pydantic import BaseModel
+from starlette.responses import JSONResponse
+
+from api.model.generic import GenericMessage, GenericExceptionMessage
+from api.model.app import App
+from api.service import manage
+
+router = APIRouter()
+
+@router.get("", responses={status.HTTP_200_OK: {"model": List[App]}})
+def list_my_apps(app_name: Optional[str] = None, status_code: Optional[int] = None,
+                 status: Optional[str] = None):
+    fields = {}
+    if app_name:
+        fields['name'] = app_name
+    if status_code:
+        fields['status_code'] = status_code
+    if status:
+        fields['status'] = status
+
+    return manage.get_my_app()

appmanage/api/v1/routers/health.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/health")
+def health_check():
+    return {"message": "StackHub API is alive"}

appmanage/cli/README.md

@@ -0,0 +1,2 @@
+# auto convert to cli
+https://www.convertapi.com/doc/cli-library

appmanage/main.py

@@ -0,0 +1,20 @@
+import logging
+
+import api.v1.api as api_router_v1
+import uvicorn
+from fastapi import FastAPI
+
+logging_format = '[%(asctime)s][%(levelname)s] %(message)s'
+logging.basicConfig(format=logging_format, level=logging.DEBUG)
+logging.getLogger().setLevel(level=logging.DEBUG)
+logging.info("Starting server")
+
+
+def get_app():
+    app = FastAPI()
+    app.include_router(api_router_v1.get_api(), prefix="/api/v1")
+    return app
+
+
+if __name__ == "__main__":
+    uvicorn.run("main:get_app", host='0.0.0.0', port=5000, reload=True)

appmanage/requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn[standard]
+gunicorn
+typer
+psutil
+pyjq