feat: full support for opendal and sync configurations between .env and docker-compose (#11754)

Signed-off-by: -LAN- <laipz8200@outlook.com>

api/extensions/storage/opendal_storage.py

@@ -1,46 +1,57 @@
+import logging
+import os
 from collections.abc import Generator
 from pathlib import Path
-from urllib.parse import urlparse
 
 import opendal
+from dotenv import dotenv_values
 
-from configs.middleware.storage.opendal_storage_config import OpenDALScheme
 from extensions.storage.base_storage import BaseStorage
 
-S3_R2_HOSTNAME = "r2.cloudflarestorage.com"
-S3_R2_COMPATIBLE_KWARGS = {
-    "delete_max_size": "700",
-    "disable_stat_with_override": "true",
-    "region": "auto",
-}
-S3_SSE_WITH_AWS_MANAGED_IAM_KWARGS = {
-    "server_side_encryption": "aws:kms",
-}
+logger = logging.getLogger(__name__)
 
 
-def is_r2_endpoint(endpoint: str) -> bool:
-    if not endpoint:
-        return False
+def _get_opendal_kwargs(*, scheme: str, env_file_path: str = ".env", prefix: str = "OPENDAL_"):
+    kwargs = {}
+    config_prefix = prefix + scheme.upper() + "_"
+    for key, value in os.environ.items():
+        if key.startswith(config_prefix):
+            kwargs[key[len(config_prefix) :].lower()] = value
 
-    parsed_url = urlparse(endpoint)
-    return bool(parsed_url.hostname and parsed_url.hostname.endswith(S3_R2_HOSTNAME))
+    file_env_vars = dotenv_values(env_file_path)
+    for key, value in file_env_vars.items():
+        if key.startswith(config_prefix) and key[len(config_prefix) :].lower() not in kwargs and value:
+            kwargs[key[len(config_prefix) :].lower()] = value
+
+    return kwargs
 
 
 class OpenDALStorage(BaseStorage):
-    def __init__(self, scheme: OpenDALScheme, **kwargs):
-        if scheme == OpenDALScheme.FS:
-            Path(kwargs["root"]).mkdir(parents=True, exist_ok=True)
+    def __init__(self, scheme: str, **kwargs):
+        kwargs = kwargs or _get_opendal_kwargs(scheme=scheme)
 
+        if scheme == "fs":
+            root = kwargs.get("root", "storage")
+            Path(root).mkdir(parents=True, exist_ok=True)
+
+        # self.op = opendal.Operator(scheme=scheme, **kwargs)
         self.op = opendal.Operator(scheme=scheme, **kwargs)
+        logger.debug(f"opendal operator created with scheme {scheme}")
+        retry_layer = opendal.layers.RetryLayer(max_times=3, factor=2.0, jitter=True)
+        self.op = self.op.layer(retry_layer)
+        logger.debug("added retry layer to opendal operator")
 
     def save(self, filename: str, data: bytes) -> None:
         self.op.write(path=filename, bs=data)
+        logger.debug(f"file {filename} saved")
 
     def load_once(self, filename: str) -> bytes:
         if not self.exists(filename):
             raise FileNotFoundError("File not found")
 
-        return self.op.read(path=filename)
+        content = self.op.read(path=filename)
+        logger.debug(f"file {filename} loaded")
+        return content
 
     def load_stream(self, filename: str) -> Generator:
         if not self.exists(filename):
@@ -50,6 +61,7 @@ class OpenDALStorage(BaseStorage):
         file = self.op.open(path=filename, mode="rb")
         while chunk := file.read(batch_size):
             yield chunk
+        logger.debug(f"file {filename} loaded as stream")
 
     def download(self, filename: str, target_filepath: str):
         if not self.exists(filename):
@@ -57,16 +69,22 @@ class OpenDALStorage(BaseStorage):
 
         with Path(target_filepath).open("wb") as f:
             f.write(self.op.read(path=filename))
+        logger.debug(f"file {filename} downloaded to {target_filepath}")
 
     def exists(self, filename: str) -> bool:
         # FIXME this is a workaround for opendal python-binding do not have a exists method and no better
         # error handler here when opendal python-binding has a exists method, we should use it
         # more https://github.com/apache/opendal/blob/main/bindings/python/src/operator.rs
         try:
-            return self.op.stat(path=filename).mode.is_file()
-        except Exception as e:
+            res = self.op.stat(path=filename).mode.is_file()
+            logger.debug(f"file {filename} checked")
+            return res
+        except Exception:
             return False
 
     def delete(self, filename: str):
         if self.exists(filename):
             self.op.delete(path=filename)
+            logger.debug(f"file {filename} deleted")
+            return
+        logger.debug(f"file {filename} not found, skip delete")