add set_permissions eu-cdse/openeo-cdse-infra#609

Author: JeroenVerstraelen

openeogeotrellis/ml/modelloader.py

@@ -15,7 +15,7 @@
 from openeo_driver.utils import generate_unique_id
 
 from openeogeotrellis.configparams import ConfigParams
-from openeogeotrellis.utils import s3_client, add_permissions
+from openeogeotrellis.utils import s3_client, set_permissions
 from .geopysparkmlmodel import GeopysparkMlModel, ModelArchitecture
 from .geopysparkcatboostmodel import GeopySparkCatBoostModel
 from .geopysparkrandomforestmodel import GeopySparkRandomForestModel
@@ -181,9 +181,9 @@ def _create_model_dir(gps_batch_jobs: 'GpsBatchJobs', use_s3: bool = False) -> s
 
             if not ml_models_dir_exists:
                 # Everyone can access the ml_models directory: `drwxrwxrwx user group`
-                add_permissions(ml_models_dir, mode=stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
+                set_permissions(ml_models_dir, mode=stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
             # Only the user has full access to this directory: `drwx------. user group`
-            add_permissions(result_dir, mode=stat.S_IRWXU, user=None)
+            set_permissions(result_dir, mode=stat.S_IRWXU, user=None)
             return str(result_dir)
 
         except Exception as e:
@@ -283,4 +283,4 @@ def _is_valid_url(url: str) -> bool:
             parsed = urlparse(url)
             return parsed.scheme in ["http", "https"] and parsed.netloc
         except Exception:
-            return False
+            return False

openeogeotrellis/utils.py

@@ -451,6 +451,25 @@ def add_permissions_with_failsafe(path: Path, mode: int, user=None, group=None):
             p.chmod(current_permission_bits | mode)
 
 
+def set_permissions(path: Path, mode: int, user=None, group=None):
+    """
+    Set permissions to a file or directory, and optionally change its ownership.
+    """
+    if str(path).lower().startswith("s3:/"):
+        logger.warning(f"set_permissions called on S3 path {path!r}, which is not supported.")
+        return
+    if not path.exists():
+        raise FileNotFoundError
+    os.chmod(path, mode)
+    if user is not None or group is not None:
+        try:
+            shutil.chown(path, user=user, group=group)
+        except LookupError as e:
+            logger.warning(f"Could not change user/group of {path} to {user}/{group}.")
+        except PermissionError as e:
+            logger.warning(f"Could not change user/group of {path} to {user}/{group}, no permissions.")
+
+
 def ensure_executor_logging(f) -> Callable:
     def setup_context_aware_logging(user_id: Optional[str], request_id: str):
         job_id = get_job_id()