[DSIP-13][python] New mechanism file plugins to Python API

dolphinscheduler-python/pydolphinscheduler/docs/source/index.rst

@@ -36,6 +36,7 @@ then go and see :doc:`tutorial` for more detail.
    cli
    config
    api
+   resources_plugin/index
 
 Indices and tables
 ==================

dolphinscheduler-python/pydolphinscheduler/docs/source/resources_plugin/develop.rst

@@ -0,0 +1,46 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+   or more contributor license agreements.  See the NOTICE file
+   distributed with this work for additional information
+   regarding copyright ownership.  The ASF licenses this file
+   to you under the Apache License, Version 2.0 (the
+   "License"); you may not use this file except in compliance
+   with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+   software distributed under the License is distributed on an
+   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+   KIND, either express or implied.  See the License for the
+   specific language governing permissions and limitations
+   under the License.
+
+How to develop
+==============
+
+When you want to create a new resource plugin, you need to add a new class in the module `resources_plugin`.
+
+The resource plug-in class needs to inherit the abstract class `ResourcePlugin` and implement its abstract method `read_file` function.
+
+The parameter of the `__init__` function of `ResourcePlugin` is the prefix of STR type. You can override this function when necessary.
+
+The `read_file` function parameter of `ResourcePlugin` is the file suffix of STR type, and its return value is the file content, if it exists and is readable.
+
+
+Example
+-------
+- Method `__init__`: Initiation method with `param`:`prefix`
+
+.. literalinclude:: ../../../src/pydolphinscheduler/resources_plugin/local.py
+    :start-after: [start init_method]
+    :end-before: [end init_method]
+
+- Method `read_file`: Get content from the given URI, The function parameter is the suffix of the file path.
+
+The file prefix has been initialized in init of the resource plug-in.
+
+The prefix plus suffix is the absolute path of the file in this resource.
+
+.. literalinclude:: ../../../src/pydolphinscheduler/resources_plugin/local.py
+    :start-after: [start read_file_method]
+    :end-before: [end read_file_method]

dolphinscheduler-python/pydolphinscheduler/docs/source/resources_plugin/index.rst

@@ -0,0 +1,28 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+   or more contributor license agreements.  See the NOTICE file
+   distributed with this work for additional information
+   regarding copyright ownership.  The ASF licenses this file
+   to you under the Apache License, Version 2.0 (the
+   "License"); you may not use this file except in compliance
+   with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+   software distributed under the License is distributed on an
+   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+   KIND, either express or implied.  See the License for the
+   specific language governing permissions and limitations
+   under the License.
+
+Resources_plugin
+================
+
+In this section
+
+.. toctree::
+   :maxdepth: 1
+
+   develop
+   resource-plugin
+   local

dolphinscheduler-python/pydolphinscheduler/docs/source/resources_plugin/local.rst

@@ -0,0 +1,32 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+   or more contributor license agreements.  See the NOTICE file
+   distributed with this work for additional information
+   regarding copyright ownership.  The ASF licenses this file
+   to you under the Apache License, Version 2.0 (the
+   "License"); you may not use this file except in compliance
+   with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+   software distributed under the License is distributed on an
+   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+   KIND, either express or implied.  See the License for the
+   specific language governing permissions and limitations
+   under the License.
+
+Local
+=====
+
+`Local` is a local resource plugin for pydolphinscheduler.
+
+When using a local resource plugin, you only need to add the `resource_plugin` parameter in the task subclass or workflow definition,
+such as `resource_plugin=Local("/tmp")`.
+
+
+For the specific use of resource plugins, you can see `How to use` in :doc:`./resource-plugin`
+
+Dive Into
+---------
+
+.. automodule:: pydolphinscheduler.resources_plugin.local

dolphinscheduler-python/pydolphinscheduler/docs/source/resources_plugin/resource-plugin.rst

@@ -0,0 +1,75 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+   or more contributor license agreements.  See the NOTICE file
+   distributed with this work for additional information
+   regarding copyright ownership.  The ASF licenses this file
+   to you under the Apache License, Version 2.0 (the
+   "License"); you may not use this file except in compliance
+   with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+   software distributed under the License is distributed on an
+   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+   KIND, either express or implied.  See the License for the
+   specific language governing permissions and limitations
+   under the License.
+
+ResourcePlugin
+==============
+
+`ResourcePlugin` is an abstract class of resource plug-in parameters of task subclass and workflow.
+All resource plugins need to inherit and override its abstract methods.
+
+Code
+----
+.. literalinclude:: ../../../src/pydolphinscheduler/core/resource_plugin.py
+   :start-after: [start resource_plugin_definition]
+   :end-before: [end resource_plugin_definition]
+
+Dive Into
+---------
+It has the following key functions.
+
+- Method `__init__`: The `__init__` function has STR type parameter `prefix`, which means the prefix of the resource.
+
+You can rewrite this function if necessary.
+
+.. literalinclude:: ../../../src/pydolphinscheduler/core/resource_plugin.py
+    :start-after: [start init_method]
+    :end-before: [end init_method]
+
+- Method `read_file`: Get content from the given URI, The function parameter is the suffix of the file path.
+
+The file prefix has been initialized in init of the resource plug-in.
+
+The prefix plus suffix is the absolute path of the file in this resource.
+
+It is an abstract function. You must rewrite it
+
+.. literalinclude:: ../../../src/pydolphinscheduler/core/resource_plugin.py
+    :start-after: [start abstractmethod read_file]
+    :end-before: [end abstractmethod read_file]
+
+.. automodule:: pydolphinscheduler.core.resource_plugin
+
+How to use
+----------
+Resource plug-ins can be used in task subclasses and workflows. You can use the resource plug-ins by adding the `resource_plugin` parameter when they are initialized.
+For example, local resource plug-ins, add `resource_plugin = Local("/tmp")`.
+
+The resource plug-ins we currently support is `local`.
+
+Here is an example.
+
+.. literalinclude:: ../../../src/pydolphinscheduler/examples/tutorial_resource_plugin.py
+   :start-after: [start workflow_declare]
+   :end-before: [end task_declare]
+
+When the resource_plugin parameter is defined in both the task subclass and the workflow, the resource_plugin defined in the task subclass is used first.
+
+If the task subclass does not define resource_plugin, but the resource_plugin is defined in the workflow, the resource_plugin in the workflow is used.
+
+Of course, if neither the task subclass nor the workflow specifies resource_plugin, the command at this time will be executed as a script,
+
+in other words, we are forward compatible.

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py

@@ -24,6 +24,7 @@
 from pydolphinscheduler import configuration
 from pydolphinscheduler.constants import TaskType
 from pydolphinscheduler.core.resource import Resource
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
 from pydolphinscheduler.exceptions import PyDSParamException, PyDSTaskNoFoundException
 from pydolphinscheduler.java_gateway import JavaGate
 from pydolphinscheduler.models import Base, Project, Tenant, User
@@ -111,6 +112,7 @@ def __init__(
         timeout: Optional[int] = 0,
         release_state: Optional[str] = configuration.WORKFLOW_RELEASE_STATE,
         param: Optional[Dict] = None,
+        resource_plugin: Optional[ResourcePlugin] = None,
         resource_list: Optional[List[Resource]] = None,
     ):
         super().__init__(name, description)
@@ -134,6 +136,7 @@ def __init__(
         self._release_state = release_state
         self.param = param
         self.tasks: dict = {}
+        self.resource_plugin = resource_plugin
         # TODO how to fix circle import
         self._task_relations: set["TaskRelation"] = set()  # noqa: F821
         self._process_definition_code = None

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/resource_plugin.py

@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""DolphinScheduler ResourcePlugin object."""
+
+
+from abc import ABCMeta, abstractmethod
+
+
+# [start resource_plugin_definition]
+class ResourcePlugin(object, metaclass=ABCMeta):
+    """ResourcePlugin object, declare resource plugin for task and workflow to dolphinscheduler.
+
+    :param prefix: A string representing the prefix of ResourcePlugin.
+
+    """
+
+    # [start init_method]
+    def __init__(self, prefix: str, *args, **kwargs):
+        self.prefix = prefix
+
+    # [end init_method]
+
+    # [start abstractmethod read_file]
+    @abstractmethod
+    def read_file(self, suf: str):
+        """Get the content of the file.
+
+        The address of the file is the prefix of the resource plugin plus the parameter suf.
+        """
+
+    # [end abstractmethod read_file]
+
+
+# [end resource_plugin_definition]

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/task.py

@@ -33,7 +33,8 @@
     ProcessDefinitionContext,
 )
 from pydolphinscheduler.core.resource import Resource
-from pydolphinscheduler.exceptions import PyDSParamException
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.exceptions import PyDSParamException, PyResPluginException
 from pydolphinscheduler.java_gateway import JavaGate
 from pydolphinscheduler.models import Base
 
@@ -101,6 +102,9 @@ class Task(Base):
 
     _task_custom_attr: set = set()
 
+    ext: set = None
+    ext_attr: str = None
+
     DEFAULT_CONDITION_RESULT = {"successNode": [""], "failedNode": [""]}
 
     def __init__(
@@ -124,6 +128,7 @@ def __init__(
         dependence: Optional[Dict] = None,
         wait_start_timeout: Optional[Dict] = None,
         condition_result: Optional[Dict] = None,
+        resource_plugin: Optional[ResourcePlugin] = None,
     ):
 
         super().__init__(name, description)
@@ -166,6 +171,8 @@ def __init__(
         self.dependence = dependence or {}
         self.wait_start_timeout = wait_start_timeout or {}
         self._condition_result = condition_result or self.DEFAULT_CONDITION_RESULT
+        self.resource_plugin = resource_plugin
+        self.get_content()
 
     @property
     def process_definition(self) -> Optional[ProcessDefinition]:
@@ -229,6 +236,44 @@ def task_params(self) -> Optional[Dict]:
         custom_attr |= self._task_custom_attr
         return self.get_define_custom(custom_attr=custom_attr)
 
+    def get_plugin(self):
+        """Return the resource plug-in.
+
+        according to parameter resource_plugin and parameter
+        process_definition.resource_plugin.
+        """
+        if self.resource_plugin is None:
+            if self.process_definition.resource_plugin is not None:
+                return self.process_definition.resource_plugin
+            else:
+                raise PyResPluginException(
+                    "The execution command of this task is a file, but the resource plugin is empty"
+                )
+        else:
+            return self.resource_plugin
+
+    def get_content(self):
+        """Get the file content according to the resource plugin."""
+        if self.ext_attr is None and self.ext is None:
+            return
+
+        _ext_attr = getattr(self, self.ext_attr)
+
+        if _ext_attr is not None:
+            if _ext_attr.endswith(tuple(self.ext)):
+                res = self.get_plugin()
+                content = res.read_file(_ext_attr)
+                setattr(self, self.ext_attr.lstrip("_"), content)
+            else:
+                index = _ext_attr.rfind(".")
+                if index != -1:
+                    raise ValueError(
+                        "This task does not support files with suffix {}, only supports {}".format(
+                            _ext_attr[index:], ",".join(str(suf) for suf in self.ext)
+                        )
+                    )
+                setattr(self, self.ext_attr.lstrip("_"), _ext_attr)
+
     def __hash__(self):
         return hash(self.code)