Added variables system
parent
c32640deab
commit
6df4788cca
|
@ -1,3 +1,4 @@
|
||||||
MODULE_RTD8_LEVELS=01234567 #number of stack level (0-7) not separated
|
MODULE_RTD8_LEVELS=01234567 #number of stack level (0-7) not separated
|
||||||
MODULE_REL4HVI4_LEVELS=01234567 #number of stack level (0-7) not separated
|
MODULE_REL4HVI4_LEVELS=01234567 #number of stack level (0-7) not separated
|
||||||
MODULE_INDUSTRIALAUTOMATION_LEVELS=01234567 #number of stack level (0-7) not separated
|
MODULE_INDUSTRIALAUTOMATION_LEVELS=01234567 #number of stack level (0-7) not separated
|
||||||
|
VARIABLES_SAVE_DIR=variables_storage #path to dir for saving variables
|
|
@ -128,6 +128,7 @@ venv/
|
||||||
ENV/
|
ENV/
|
||||||
env.bak/
|
env.bak/
|
||||||
venv.bak/
|
venv.bak/
|
||||||
|
variables_storage/
|
||||||
|
|
||||||
# Spyder project settings
|
# Spyder project settings
|
||||||
.spyderproject
|
.spyderproject
|
||||||
|
|
|
@ -5,10 +5,11 @@ load_dotenv()
|
||||||
|
|
||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
from src.modules.router import router as modulesrouter
|
from src.modules.router import router as modulesrouter
|
||||||
|
from src.variables.router import router as variablesrouter
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
app = FastAPI(root_path="/api")
|
app = FastAPI(root_path="/api")
|
||||||
|
|
||||||
app.include_router(router=modulesrouter)
|
app.include_router(router=modulesrouter)
|
||||||
|
app.include_router(router=variablesrouter)
|
||||||
|
|
|
@ -0,0 +1,7 @@
|
||||||
|
from pydantic import BaseModel
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
|
class SetVariable(BaseModel):
|
||||||
|
data: Any
|
||||||
|
default: bool
|
|
@ -0,0 +1,26 @@
|
||||||
|
from fastapi import APIRouter, status, HTTPException
|
||||||
|
import src.variables.service as service
|
||||||
|
from src.variables.models import SetVariable
|
||||||
|
|
||||||
|
|
||||||
|
router = APIRouter(prefix="/variables", tags=["Variables system"])
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/{variable}", description="Set variable")
|
||||||
|
def set_variable(variable: str, params: SetVariable):
|
||||||
|
service.set_variable(variable=variable, data=params.data, default=params.default)
|
||||||
|
|
||||||
|
@router.get("/{variable}", description="Get variable data", response_model=str)
|
||||||
|
def get_variable(variable: str):
|
||||||
|
try:
|
||||||
|
return service.get_variable(variable=variable)
|
||||||
|
except:
|
||||||
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
||||||
|
|
||||||
|
@router.get("/", description="Get variables list", response_model=list)
|
||||||
|
def get_variables():
|
||||||
|
return service.variables.keys()
|
||||||
|
|
||||||
|
@router.delete("/{variable}", description="Remove variable")
|
||||||
|
def delete_variable(variable: str):
|
||||||
|
service.remove_variable(variable=variable)
|
|
@ -0,0 +1,67 @@
|
||||||
|
import os, re
|
||||||
|
|
||||||
|
|
||||||
|
variables: dict = {}
|
||||||
|
|
||||||
|
|
||||||
|
def __save_data(variable, data):
|
||||||
|
with open(variable, "w") as file:
|
||||||
|
file.write(data)
|
||||||
|
|
||||||
|
|
||||||
|
def __load_data(variable, type):
|
||||||
|
with open(variable, "r") as file:
|
||||||
|
return file.read()
|
||||||
|
|
||||||
|
def __create_file_path(variable: str):
|
||||||
|
default_path: str = os.getenv("VARIABLES_SAVE_DIR", "variables_storage")
|
||||||
|
if not os.path.exists(default_path):
|
||||||
|
os.makedirs(default_path)
|
||||||
|
return f"{default_path}/{re.sub(r'[^\w\-_.]', '_', variable)}"
|
||||||
|
|
||||||
|
|
||||||
|
def set_variable(variable: str, data, default: bool):
|
||||||
|
"""
|
||||||
|
Set variable
|
||||||
|
"""
|
||||||
|
variables[variable] = data
|
||||||
|
|
||||||
|
if default:
|
||||||
|
__save_data(variable=__create_file_path(variable), data=data)
|
||||||
|
|
||||||
|
|
||||||
|
def load_variable(variable: str) -> str:
|
||||||
|
"""
|
||||||
|
Load variable from File if exist
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return __load_data(variable=__create_file_path(variable=variable))
|
||||||
|
except FileNotFoundError:
|
||||||
|
raise FileNotFoundError(f"Unnable find variable: {variable}")
|
||||||
|
except:
|
||||||
|
raise ValueError((f"Error with loading variable: {variable}"))
|
||||||
|
|
||||||
|
|
||||||
|
def get_variable(variable: str) -> str | None:
|
||||||
|
"""
|
||||||
|
Get variable if exist
|
||||||
|
"""
|
||||||
|
if variable in variables:
|
||||||
|
return variables[variable]
|
||||||
|
else:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def remove_variable(variable: str):
|
||||||
|
if variable in variables:
|
||||||
|
variables.pop(variable)
|
||||||
|
|
||||||
|
filePath = __create_file_path(variable)
|
||||||
|
if os.path.exists(filePath):
|
||||||
|
os.remove(filePath)
|
||||||
|
|
||||||
|
|
||||||
|
#autoload variables from files
|
||||||
|
__create_file_path("none")
|
||||||
|
for file in os.listdir(os.getenv("VARIABLES_SAVE_DIR", "variables_storage")):
|
||||||
|
variables[file] = load_variable(file)
|
Loading…
Reference in New Issue