Add opendal provider

providers/common/opendal/src/airflow/providers/common/opendal/connections/connection_parser.py

@@ -37,7 +37,7 @@ def parse(self, conn: Connection) -> dict[str, Any]:
         })
 
         params = {k: v for k, v in params.items() if v is not None}
-        return {"scheme": "s3", "params": params}
+        return {"scheme": "s3",  **params}
 
 
 class OpenDALOperatorFactory:
@@ -61,14 +61,14 @@ def get_opendal_operator_args(self, conn: Connection, opendal_config_operator_ar
         op_args = {}
         for parser in self._connection_parsers:
             if conn_scheme in parser.airflow_conn_type:
-                parsed_conn = parser.parse(conn)
-                params = parsed_conn.get("params")
+                params = parser.parse(conn)
+                break
 
         if config_type == "destination":
             op_args = conn.extra_dejson.get("destination_config", {}).get("operator_args", {})
 
         if config_type == "source":
             op_args = conn.extra_dejson.get("source_config", {}).get("operator_args", {})
 
-
+        print(f"params: {params}")
         return {**params, **op_args, **opendal_config_operator_args}

providers/common/opendal/src/airflow/providers/common/opendal/filesystem/opendal_fs.py

@@ -8,16 +8,19 @@ class SourceConfig(BaseModel):
     operator_args: Optional[dict[str, str]] = None
     path: str = None
 
+
 class DestinationConfig(BaseModel):
     conn_id: str = None
     operator_args: Optional[dict[str, str]] = None
     path: str = None
 
+
 class OpenDALConfig(BaseModel):
     action: Literal["read", "write", "copy"]
     source_config: SourceConfig
     destination_config: Optional[DestinationConfig] = None
 
+
 class OpenDALBaseFileSystem:
     def __init__(self,
                  opendal_config: OpenDALConfig,
@@ -42,27 +45,15 @@ def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
     def execute_opendal_task(self):
-        return self.source_operator.read(self.opendal_config.source_config.path)
+        return self.source_operator.read(self.opendal_config.get("source_config",{}).get("path", "/")).decode("utf-8")
 
 
 class OpenDALWrite(OpenDALBaseFileSystem):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
     def execute_opendal_task(self):
-        return self.source_operator.write(self.opendal_config.source_config.path, self.data)
-
-
-
-'''
-config:
-    action: read/write/copy
-    source_config:
-        conn_id
-        operator_args:
-        path
-    destination_config:
-        conn_id
-        operator_args:
-        path
-'''
+        if self.data and isinstance(self.data, str):
+            self.data = self.data.encode("utf-8")
+
+        return self.source_operator.write(self.opendal_config.get("source_config", {}).get("path"), self.data)

providers/common/opendal/src/airflow/providers/common/opendal/hooks/opendal.py

@@ -34,6 +34,7 @@ def get_operator(self) -> Operator:
             self.config.get("operator_args"),
             self.config_type,
         )
+        print(f"op_args: {op_args}")
         return Operator(**op_args)
 
     @cached_property

providers/common/opendal/src/airflow/providers/common/opendal/operators/opendal.py

@@ -1,3 +1,4 @@
+import importlib
 from typing import Any
 
 from airflow.hooks.base import BaseHook
@@ -23,16 +24,31 @@ class OpenDALTaskOperator(BaseOperator):
     def __init__(self,
                  *,
                  opendal_config: OpenDALConfig,
+                 data: str | bytes = None,
                  **kwargs
                  ):
         super().__init__(**kwargs)
         self.opendal_config = opendal_config
+        self.data = data
 
     def execute(self, context: Context) -> Any:
-        action = self.opendal_config.action
 
-        source_operator = self.hook(self.opendal_config.source_config).get_operator
-        destination_operator = self.hook(self.opendal_config.destination_config, "destination").get_operator
+        action = self.opendal_config.get("action")
+
+        source_operator = self.hook(self.opendal_config.get("source_config")).get_operator
+        destination_operator = self.hook(self.opendal_config.get("destination_config"), "destination").get_operator if self.opendal_config.get("destination_config") else None
+
+        module = importlib.import_module(f"airflow.providers.common.opendal.filesystem.opendal_fs")
+        operator_class = getattr(module, f"OpenDAL{action.capitalize()}")
+
+        opendal_operator = operator_class(
+            opendal_config=self.opendal_config,
+            source_operator=source_operator,
+            destination_operator=destination_operator,
+            data=self.data,
+        )
+
+        return opendal_operator.execute_opendal_task()