fix(Config): refactor project/data_folder_path

requirements.txt

@@ -60,4 +60,3 @@ werkzeug==3.1.8
     #   flask
     #   flask-cors
 
-opengeodeweb-microservice==1.*,>=1.1.3

src/opengeodeweb_back/app.py

@@ -18,15 +18,6 @@
 def create_app(name: str) -> flask.Flask:
     app = flask.Flask(name)
 
-    """ Config variables """
-    FLASK_DEBUG = (
-        True if os.environ.get("FLASK_DEBUG", default=None) == "True" else False
-    )
-    if FLASK_DEBUG == False:
-        app.config.from_object(app_config.ProdConfig)
-    else:
-        app.config.from_object(app_config.DevConfig)
-
     @app.before_request
     def before_request() -> flask.Response | None:
         if flask.request.method == "OPTIONS":
@@ -102,14 +93,12 @@ def run_server(app: Flask) -> None:
     parser.add_argument(
         "--host",
         type=str,
-        default=app.config.get("DEFAULT_HOST"),
         help="Host to run on",
     )
     parser.add_argument(
         "-p",
         "--port",
-        type=int,
-        default=app.config.get("DEFAULT_PORT"),
+        type=str,
         help="Port to listen on",
     )
     parser.add_argument(
@@ -119,55 +108,102 @@ def run_server(app: Flask) -> None:
         help="Whether to run in debug mode",
         action="store_true",
     )
+    parser.add_argument(
+        "-pfp",
+        "--project_folder_path",
+        type=str,
+        help="Path to the folder where the project is stored",
+    )
     parser.add_argument(
         "-dfp",
         "--data_folder_path",
         type=str,
-        default=app.config.get("DEFAULT_DATA_FOLDER_PATH"),
-        help="Path to the folder where data is stored",
+        help="Path to the folder where the data is stored",
     )
     parser.add_argument(
         "-ufp",
         "--upload_folder_path",
         type=str,
-        default=app.config.get("UPLOAD_FOLDER"),
         help="Path to the folder where uploads are stored",
     )
     parser.add_argument(
         "-origins",
         "--allowed_origins",
         nargs="+",
-        default=app.config.get("ORIGINS"),
         help="Origins that are allowed to connect to the server",
     )
     parser.add_argument(
         "-t",
         "--timeout",
-        default=app.config.get("MINUTES_BEFORE_TIMEOUT"),
         help="Number of minutes before the server times out",
     )
     args, _ = parser.parse_known_args()
-    app.config.update(DATA_FOLDER_PATH=args.data_folder_path)
-    app.config.update(
-        EXTENSIONS_FOLDER_PATH=os.path.join(str(args.data_folder_path), "extensions")
-    )
-    app.config.update(UPLOAD_FOLDER=args.upload_folder_path)
-    app.config.update(MINUTES_BEFORE_TIMEOUT=args.timeout)
-    flask_cors.CORS(app, origins=args.allowed_origins)
+
+    if args.project_folder_path is None:
+        raise ValueError("project_folder_path must be provided")
+    else:
+        args.project_folder_path = os.path.abspath(args.project_folder_path)
+
+    if args.debug:
+        app.config.from_object(app_config.DevConfig(args.project_folder_path))
+    else:
+        app.config.from_object(app_config.ProdConfig(args.project_folder_path))
+
+    if args.host is not None:
+        app.config.update(HOST=args.host)
+    else:
+        args.host = app.config.get("HOST")
+
+    if args.port is not None:
+        app.config.update(PORT=args.port)
+    else:
+        args.port = app.config.get("PORT")
+
+    if args.debug is not None:
+        app.config.update(FLASK_DEBUG=args.debug)
+    else:
+        args.debug = app.config.get("FLASK_DEBUG")
+
+    if args.data_folder_path is not None:
+        app.config.update(DATA_FOLDER_PATH=args.data_folder_path)
+    else:
+        args.data_folder_path = app.config.get("DATA_FOLDER_PATH")
+
+    if args.upload_folder_path is not None:
+        app.config.update(UPLOAD_FOLDER_PATH=args.upload_folder_path)
+    else:
+        args.upload_folder_path = app.config.get("UPLOAD_FOLDER_PATH")
+
+    if args.allowed_origins is not None:
+        app.config.update(ALLOWED_ORIGINS=args.allowed_origins)
+    else:
+        args.allowed_origins = app.config.get("ALLOWED_ORIGINS")
+
+    if args.timeout is not None:
+        app.config.update(MINUTES_BEFORE_TIMEOUT=args.timeout)
+    else:
+        args.timeout = app.config.get("MINUTES_BEFORE_TIMEOUT")
+
     print(f"{args=}", flush=True)
 
-    db_filename: str = app.config.get("DATABASE_FILENAME") or "project.db"
-    db_path = os.path.join(str(args.data_folder_path), db_filename)
+    db_filename = app.config.get("DATABASE_FILENAME")
+    if not isinstance(db_filename, str):
+        raise TypeError(
+            f"DATABASE_FILENAME config must be a string, got {db_filename!r}"
+        )
+    db_path = os.path.join(str(app.config.get("DATA_FOLDER_PATH")), db_filename)
     os.makedirs(os.path.dirname(db_path), exist_ok=True)
     app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
 
     connection.init_database(db_path)
     print(f"Database initialized at: {db_path}", flush=True)
+
+    flask_cors.CORS(app, origins=args.allowed_origins)
     app.run(
-        debug=args.debug,
-        host=args.host,
-        port=args.port,
+        debug=app.config.get("FLASK_DEBUG"),
+        host=app.config.get("HOST"),
+        port=app.config.get("PORT"),
         ssl_context=app.config.get("SSL"),
     )
 

src/opengeodeweb_back/app_config.py

@@ -5,33 +5,35 @@
 # Third party imports
 # Local application imports
 
+base_dir = os.path.dirname(os.path.abspath(__file__))
+
 
 class Config(object):
     FLASK_DEBUG = os.environ.get("FLASK_DEBUG", default=False)
-    DEFAULT_HOST = "localhost"
-    DEFAULT_PORT = "5000"
+    HOST = "localhost"
+    PORT = "5000"
     CORS_HEADERS = "Content-Type"
-    UPLOAD_FOLDER = "./uploads"
     REQUEST_COUNTER = 0
     LAST_REQUEST_TIME = time.time()
     LAST_PING_TIME = time.time()
     DATABASE_FILENAME = "project.db"
 
+    def __init__(self, project_folder_path: str):
+        self.PROJECT_FOLDER_PATH = project_folder_path
+        self.DATA_FOLDER_PATH = os.path.join(project_folder_path, "data")
+        self.EXTENSIONS_FOLDER_PATH = os.path.join(project_folder_path, "extensions")
+        self.UPLOAD_FOLDER_PATH = os.path.join(project_folder_path, "uploads")
+
 
 class ProdConfig(Config):
     SSL = None
     ORIGINS = ""
     MINUTES_BEFORE_TIMEOUT = "1"
     SECONDS_BETWEEN_SHUTDOWNS = "10"
-    DATA_FOLDER_PATH = "/data"
-    EXTENSIONS_FOLDER_PATH = os.path.join(DATA_FOLDER_PATH, "extensions")
 
 
 class DevConfig(Config):
     SSL = None
     ORIGINS = "*"
     MINUTES_BEFORE_TIMEOUT = "1"
     SECONDS_BETWEEN_SHUTDOWNS = "10"
-    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-    DATA_FOLDER_PATH = os.path.join(BASE_DIR, "data")
-    EXTENSIONS_FOLDER_PATH = os.path.join(DATA_FOLDER_PATH, "extensions")

src/opengeodeweb_back/geode_functions.py

@@ -46,7 +46,7 @@ def get_data_info(data_id: str) -> Data:
 
 
 def upload_file_path(filename: str) -> str:
-    upload_folder = flask.current_app.config["UPLOAD_FOLDER"]
+    upload_folder = flask.current_app.config["UPLOAD_FOLDER_PATH"]
     secure_filename = werkzeug.utils.secure_filename(filename)
     return os.path.abspath(os.path.join(upload_folder, secure_filename))
 

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -66,20 +66,22 @@ def allowed_files() -> flask.Response:
     methods=schemas_dict["upload_file"]["methods"],
 )
 def upload_file() -> flask.Response:
-    UPLOAD_FOLDER = flask.current_app.config["UPLOAD_FOLDER"]
-    print(f"{UPLOAD_FOLDER=}", flush=True)
-    if not os.path.exists(UPLOAD_FOLDER):
-        os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+    UPLOAD_FOLDER_PATH = flask.current_app.config["UPLOAD_FOLDER_PATH"]
+    print(f"{UPLOAD_FOLDER_PATH=}", flush=True)
+    if not os.path.exists(UPLOAD_FOLDER_PATH):
+        os.makedirs(UPLOAD_FOLDER_PATH, exist_ok=True)
 
     file = flask.request.files["file"]
     if file.filename is None:
         flask.abort(400, "Filename is required")
     filename = werkzeug.utils.secure_filename(os.path.basename(file.filename))
     print(f"{filename=}", flush=True)
-    file_path = os.path.join(UPLOAD_FOLDER, filename)
+    file_path = os.path.join(UPLOAD_FOLDER_PATH, filename)
     file.save(file_path)
     if filename.lower().endswith(".csv.json"):
-        shutil.copyfile(file_path, os.path.join(UPLOAD_FOLDER, filename[:-9] + ".json"))
+        shutil.copyfile(
+            file_path, os.path.join(UPLOAD_FOLDER_PATH, filename[:-9] + ".json")
+        )
     return flask.make_response({"message": "File uploaded"}, 201)
 
 
@@ -585,8 +587,13 @@ def import_project() -> flask.Response:
 
     try:
         if os.path.exists(data_folder_path):
-            shutil.rmtree(data_folder_path)
-        os.makedirs(data_folder_path, exist_ok=True)
+            for item in os.scandir(data_folder_path):
+                if item.is_dir(follow_symlinks=False):
+                    shutil.rmtree(item.path)
+                else:
+                    os.remove(item.path)
+        else:
+            os.makedirs(data_folder_path, exist_ok=True)
     except PermissionError:
         flask.abort(423, "Project files are locked; cannot overwrite")
 

tests/conftest.py

@@ -22,16 +22,17 @@
 
 @pytest.fixture(scope="session", autouse=True)
 def configure_test_environment() -> Generator[None, None, None]:
-    base_path = Path(__file__).parent
+    base_path = Path(__file__).parent.absolute()
     test_data_path = base_path / "data"
 
     shutil.rmtree("./data", ignore_errors=True)
     shutil.copytree(test_data_path, f"./data/{TEST_ID}/", dirs_exist_ok=True)
 
     app.config["TESTING"] = True
     app.config["SERVER_NAME"] = "TEST"
+    app.config["PROJECT_FOLDER_PATH"] = base_path
     app.config["DATA_FOLDER_PATH"] = "./data/"
-    app.config["UPLOAD_FOLDER"] = "./tests/data/"
+    app.config["UPLOAD_FOLDER_PATH"] = "./tests/data/"
 
     db_path = os.path.join(base_path, "data", "project.db")
     app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"

tests/test_utils_functions.py

@@ -232,21 +232,21 @@ def test_generate_native_viewable_and_light_viewable_from_file_with_multi_dots(
 def test_send_file_multiple_returns_zip(client: FlaskClient, tmp_path: Path) -> None:
     app = client.application
     with app.app_context():
-        app.config["UPLOAD_FOLDER"] = str(tmp_path)
+        app.config["UPLOAD_FOLDER_PATH"] = str(tmp_path)
         file_paths = []
         for i, content in [(1, b"hello 1"), (2, b"hello 2")]:
             file_path = tmp_path / f"tmp_send_file_{i}.txt"
             file_path.write_bytes(content)
             file_paths.append(str(file_path))
         with app.test_request_context():
             response = utils_functions.send_file(
-                app.config["UPLOAD_FOLDER"], file_paths, "bundle"
+                app.config["UPLOAD_FOLDER_PATH"], file_paths, "bundle"
             )
             assert response.status_code == 200
             assert response.mimetype == "application/zip"
             new_file_name = response.headers.get("new-file-name")
             assert new_file_name == "bundle.zip"
-            zip_path = os.path.join(app.config["UPLOAD_FOLDER"], new_file_name)
+            zip_path = os.path.join(app.config["UPLOAD_FOLDER_PATH"], new_file_name)
             with zipfile.ZipFile(zip_path, "r") as zip_file:
                 zip_entries = zip_file.namelist()
                 assert "tmp_send_file_1.txt" in zip_entries
@@ -259,18 +259,20 @@ def test_send_file_single_returns_octet_binary(
 ) -> None:
     app = client.application
     with app.app_context():
-        app.config["UPLOAD_FOLDER"] = str(tmp_path)
+        app.config["UPLOAD_FOLDER_PATH"] = str(tmp_path)
         file_path = tmp_path / "tmp_send_file_1.txt"
         file_path.write_bytes(b"hello 1")
         with app.test_request_context():
             response = utils_functions.send_file(
-                app.config["UPLOAD_FOLDER"], [str(file_path)], "tmp_send_file_1.txt"
+                app.config["UPLOAD_FOLDER_PATH"],
+                [str(file_path)],
+                "tmp_send_file_1.txt",
             )
             assert response.status_code == 200
             assert response.mimetype == "application/octet-binary"
             new_file_name = response.headers.get("new-file-name")
             assert new_file_name == "tmp_send_file_1.txt"
-            zip_path = os.path.join(app.config["UPLOAD_FOLDER"], new_file_name)
+            zip_path = os.path.join(app.config["UPLOAD_FOLDER_PATH"], new_file_name)
             with open(zip_path, "rb") as f:
                 file_bytes = f.read()
             assert file_bytes == b"hello 1"