-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
75 lines (62 loc) · 2.64 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import os
import argparse
import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from modules.loaders import NacosJsonSchemaLoader as loader
from modules.loggers import system_log
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
class ConfigData(BaseModel):
schema_title: str
schema_version: str
description: str
type: str
properties: list
@app.get("/load_config", response_class=JSONResponse)
def load_config():
try:
return loader().download_from_nacos()
except Exception as e:
system_log.error(f"Configuration file not found: {e!s}")
return JSONResponse(
content={"error": "Configuration file not found"}, status_code=404
)
@app.post("/save_config", response_class=JSONResponse)
def save_config(config_data: ConfigData):
try:
loader().upload_to_nacos(config_data.model_dump())
return {"message": "Configuration saved successfully"}
except Exception as e:
system_log.error(f"Configuration save failed: {e!s}")
return {"message": "Configuration save failed"}
@app.get("/", response_class=HTMLResponse)
async def index():
with open("static/index.html", encoding="utf-8") as file:
return HTMLResponse(content=file.read())
def set_system_environment(server_args) -> None:
os.environ["env"] = server_args.env
os.environ["nacos_addr"] = server_args.nacos_addr
os.environ["nacos_group"] = server_args.nacos_group
os.environ["nacos_data_id"] = server_args.nacos_data_id
os.environ["nacos_user"] = server_args.nacos_user
os.environ["nacos_password"] = server_args.nacos_password
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--env", type=str, required=False, help="environment name.")
parser.add_argument("--address", type=str, required=True, help="server address")
parser.add_argument("--port", type=str, default="8000", help="server port")
parser.add_argument("--nacos_addr", type=str, required=True, help="nacos host")
parser.add_argument(
"--nacos_group", type=str, default="DEFAULT_GROUP", help="nacos group"
)
parser.add_argument("--nacos_data_id", type=str, required=True, help="data id.")
parser.add_argument("--nacos_user", type=str, default="nacos", help="nacos user")
parser.add_argument(
"--nacos_password", type=str, default="nacos", help="nacos password"
)
args = parser.parse_args()
set_system_environment(args)
uvicorn.run(app, host=args.address, port=int(args.port))