You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zh...@apache.org on 2022/11/09 09:46:30 UTC
[dolphinscheduler-sdk-python] branch main updated: [chore] Migrate code from main repo apache/dolphinscheduler (#1)
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler-sdk-python.git
The following commit(s) were added to refs/heads/main by this push:
new 05f87a9 [chore] Migrate code from main repo apache/dolphinscheduler (#1)
05f87a9 is described below
commit 05f87a99b1e007733ee83cf71d8768e404cfee9e
Author: Jay Chung <zh...@gmail.com>
AuthorDate: Wed Nov 9 17:46:25 2022 +0800
[chore] Migrate code from main repo apache/dolphinscheduler (#1)
Currently, our Python API code is a module in apache/dolphinscheduler codebase,
each time users change Python API code, they need to run all requests CI check
for dolphinscheduler and Python API, But if the user does only change Python
code, it could be merged if Python API CI pass and do not dependent on others CI.
Besides, we release Python API as the same version of dolphinscheduler. It is
easy for user to match Python API version. But when Python API does not change
any code, but dolphinscheduler release a bugfix version, Python API has to
release the new version to match dolphinscheduler. This happened when we
released Python API 2.0.6 and 2.0.7. 2.0.6 and 2.0.7 is bugfix version, and
Python API does not change any code, so the PyPI package is the same.
Separate Python API also makes our code more sense, we will have more
distinguished code in dolphinscheduler and Python API new repository.
Have separate issue tracker and changelog for information to users.
see more detail in mail thread: https://lists.apache.org/thread/4z7l5l54c4d81smjlk1n8nq380p9f0oo
---
.dlc.json | 25 +++
.flake8 | 1 -
.github/PULL_REQUEST_TEMPLATE.md | 13 ++
.github/workflows/ci.yaml | 151 +++++++++++++++-
.gitignore | 15 ++
.licenserc.yaml | 30 ++++
.pre-commit-config.yaml | 17 +-
DEVELOP.md | 15 +-
README.md | 4 +-
RELEASE.md | 2 +
UPDATING.md | 1 +
docs/source/conf.py | 2 +-
docs/source/index.rst | 1 +
docs/source/resources_plugin/develop.rst | 46 +++++
.../{index.rst => resources_plugin/github.rst} | 34 ++--
docs/source/resources_plugin/gitlab.rst | 46 +++++
docs/source/{ => resources_plugin}/index.rst | 37 ++--
.../{index.rst => resources_plugin/local.rst} | 31 +---
docs/source/resources_plugin/oss.rst | 44 +++++
docs/source/resources_plugin/resource-plugin.rst | 75 ++++++++
docs/source/{index.rst => resources_plugin/s3.rst} | 33 ++--
docs/source/start.rst | 10 +-
docs/source/tutorial.rst | 2 +-
examples/yaml_define/Spark.yaml | 1 -
setup.py | 12 +-
src/pydolphinscheduler/constants.py | 9 +
src/pydolphinscheduler/core/base.py | 74 --------
src/pydolphinscheduler/core/base_side.py | 40 -----
src/pydolphinscheduler/core/configuration.py | 193 --------------------
src/pydolphinscheduler/core/process_definition.py | 3 +
src/pydolphinscheduler/core/resource_plugin.py | 58 ++++++
src/pydolphinscheduler/core/task.py | 52 +++++-
.../examples/tutorial_resource_plugin.py | 64 +++++++
src/pydolphinscheduler/exceptions.py | 4 +
src/pydolphinscheduler/java_gateway.py | 87 ++++++++-
src/pydolphinscheduler/models/base_side.py | 8 +
src/pydolphinscheduler/models/project.py | 31 ++++
src/pydolphinscheduler/models/tenant.py | 38 +++-
src/pydolphinscheduler/models/user.py | 55 +++++-
.../{side => resources_plugin}/__init__.py | 21 +--
.../base/__init__.py} | 14 +-
.../resources_plugin/base/bucket.py | 86 +++++++++
.../resources_plugin/base/git.py | 115 ++++++++++++
src/pydolphinscheduler/resources_plugin/github.py | 106 +++++++++++
src/pydolphinscheduler/resources_plugin/gitlab.py | 112 ++++++++++++
src/pydolphinscheduler/resources_plugin/local.py | 56 ++++++
src/pydolphinscheduler/resources_plugin/oss.py | 76 ++++++++
src/pydolphinscheduler/resources_plugin/s3.py | 74 ++++++++
src/pydolphinscheduler/side/project.py | 42 -----
src/pydolphinscheduler/side/queue.py | 42 -----
src/pydolphinscheduler/side/tenant.py | 45 -----
src/pydolphinscheduler/side/user.py | 79 ---------
src/pydolphinscheduler/tasks/datax.py | 10 +-
src/pydolphinscheduler/tasks/python.py | 43 ++---
src/pydolphinscheduler/tasks/shell.py | 5 +-
src/pydolphinscheduler/tasks/spark.py | 10 --
src/pydolphinscheduler/tasks/sql.py | 7 +-
tests/core/test_task.py | 117 ++++++++++++-
tests/example/test_example.py | 6 +-
tests/integration/conftest.py | 2 +-
tests/integration/test_project.py | 78 +++++++++
tests/integration/test_tenant.py | 86 +++++++++
tests/integration/test_user.py | 107 +++++++++++
.../resources_plugin/__init__.py | 14 +-
tests/resources_plugin/test_github.py | 195 +++++++++++++++++++++
tests/resources_plugin/test_gitlab.py | 116 ++++++++++++
tests/resources_plugin/test_local.py | 108 ++++++++++++
tests/resources_plugin/test_oss.py | 112 ++++++++++++
tests/resources_plugin/test_resource_plugin.py | 75 ++++++++
tests/resources_plugin/test_s3.py | 79 +++++++++
tests/tasks/test_datax.py | 91 +++++++++-
tests/tasks/test_python.py | 58 +++++-
tests/tasks/test_shell.py | 43 ++++-
tests/tasks/test_spark.py | 3 +-
tests/tasks/test_sql.py | 45 ++++-
tests/test_java_gateway.py | 52 ------
76 files changed, 2912 insertions(+), 782 deletions(-)
diff --git a/.dlc.json b/.dlc.json
new file mode 100644
index 0000000..bd24e4c
--- /dev/null
+++ b/.dlc.json
@@ -0,0 +1,25 @@
+{
+ "ignorePatterns": [
+ {
+ "pattern": "^http://localhost"
+ },
+ {
+ "pattern": "^https://img.shields.io/badge"
+ }
+ ],
+ "httpHeaders": [
+ {
+ "urls": ["https://docs.github.com/"],
+ "headers": {
+ "Accept-Encoding": "zstd, br, gzip, deflate"
+ }
+ }
+ ],
+ "timeout": "10s",
+ "retryOn429": true,
+ "retryCount": 10,
+ "fallbackRetryDelay": "1000s",
+ "aliveStatusCodes": [
+ 200
+ ]
+}
diff --git a/.flake8 b/.flake8
index 120b42f..4a1df20 100644
--- a/.flake8
+++ b/.flake8
@@ -28,7 +28,6 @@ exclude =
dist,
htmlcov,
.tox,
- dist,
ignore =
# It's clear and not need to add docstring
D107, # D107: Don't require docstrings on __init__
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
new file mode 100644
index 0000000..ea422e3
--- /dev/null
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -0,0 +1,13 @@
+<!--Thanks for you contribute to Apache DolphinScheduler Python API, You can see more detail about contributing in https://github.com/apache/dolphinscheduler-sdk-python/DEVELOP.md .-->
+
+## Brief Summary of The Change
+
+<!--Please include `fixes: #XXXX(ISSUE_NUMBER)` to automatically close any corresponding issue when the pull request is merged. Alternatively if not fully closed you can say `related: #XXXX(ISSUE_NUMBER)`.-->
+
+## Pull Request checklist
+
+I confirm that the following checklist has been completed.
+
+- [ ] Add/Change **test cases** for the changes.
+- [ ] Add/Change the related **documentation**.
+- [ ] (Optional) Add your change to `UPDATING.md` when it is an incompatible change.
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 9f660c1..f6b0beb 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -30,11 +30,160 @@ env:
DEPENDENCES: pip setuptools wheel tox
jobs:
+ license:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: apache/skywalking-eyes/header@main
+ dead-link:
+ runs-on: ubuntu-latest
+ needs: license
+ timeout-minutes: 30
+ steps:
+ - uses: actions/checkout@v3
+ - run: sudo npm install -g markdown-link-check@3.10.0
+ - run: |
+ for file in $(find . -name "*.md"); do
+ markdown-link-check -c .dlc.json -q "$file"
+ done
+ lint:
+ timeout-minutes: 15
+ runs-on: ubuntu-latest
+ needs: license
+ steps:
+ - uses: actions/checkout@v3
+ - name: Set up Python 3.7
+ uses: actions/setup-python@v4
+ with:
+ python-version: 3.7
+ - name: Install Dependences
+ run: |
+ python -m pip install --upgrade ${{ env.DEPENDENCES }}
+ - name: Run All Lint Check
+ run: |
+ python -m tox -vv -e lint
+ pytest:
+ timeout-minutes: 15
+ needs: lint
+ runs-on: ${{ matrix.os }}
+ strategy:
+ fail-fast: false
+ matrix:
+ # YAML parse `3.10` to `3.1`, so we have to add quotes for `'3.10'`, see also:
+ # https://github.com/actions/setup-python/issues/160#issuecomment-724485470
+ python-version: [3.6, 3.7, 3.8, 3.9, '3.10', 3.11-dev]
+ os: [ubuntu-latest, macOS-latest, windows-latest]
+ # Skip because dependence [py4j](https://pypi.org/project/py4j/) not work on those environments
+ exclude:
+ - os: windows-latest
+ python-version: '3.10'
+ - os: windows-latest
+ python-version: 3.11-dev
+ steps:
+ - uses: actions/checkout@v3
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@v4
+ with:
+ python-version: ${{ matrix.python-version }}
+ - name: Install Dependences
+ run: |
+ python -m pip install --upgrade ${{ env.DEPENDENCES }}
+ - name: Run All Tests
+ run: |
+ python -m tox -vv -e code-test
+ doc-build:
+ timeout-minutes: 15
+ needs: lint
+ runs-on: ubuntu-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ env-list: [doc-build, doc-build-multi]
+ steps:
+ - uses: actions/checkout@v3
+ - name: Set up Python 3.7
+ uses: actions/setup-python@v4
+ with:
+ python-version: 3.7
+ - name: Install Dependences
+ run: |
+ python -m pip install --upgrade ${{ env.DEPENDENCES }}
+ - name: Run Build Docs Tests ${{ matrix.env-list }}
+ run: |
+ python -m tox -vv -e ${{ matrix.env-list }}
+ local-ci:
+ timeout-minutes: 15
+ needs:
+ - pytest
+ - doc-build
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v3
+ - name: Set up Python 3.7
+ uses: actions/setup-python@v4
+ with:
+ python-version: 3.7
+ - name: Install Dependences
+ run: |
+ python -m pip install --upgrade ${{ env.DEPENDENCES }}
+ - name: Run Tests Build Docs
+ run: |
+ python -m tox -vv -e local-ci
+ integrate-test:
+ needs: license
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ steps:
+ - name: Checkout Dolphinscheduler SDK Python
+ uses: actions/checkout@v3
+ with:
+ path: src
+ - name: Checkout Dolphinscheduler
+ uses: actions/checkout@v3
+ with:
+ repository: apache/dolphinscheduler
+ path: dolphinscheduler
+ submodules: true
+ - name: Cache local Maven repository
+ uses: actions/cache@v3
+ with:
+ path: ~/.m2/repository
+ key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
+ restore-keys: ${{ runner.os }}-maven-
+ # Switch to project root directory to run mvnw command
+ - name: Build Image
+ working-directory: dolphinscheduler
+ run: |
+ ./mvnw -B clean install \
+ -Dmaven.test.skip \
+ -Dmaven.javadoc.skip \
+ -Dcheckstyle.skip=true \
+ -Pdocker,release -Ddocker.tag=ci \
+ -pl dolphinscheduler-standalone-server -am
+ - name: Set up Python 3.7
+ uses: actions/setup-python@v4
+ with:
+ python-version: 3.7
+ - name: Install Dependences
+ run: |
+ python -m pip install --upgrade ${{ env.DEPENDENCES }}
+ - name: Run Integrate Tests
+ working-directory: src
+ run: |
+ python -m tox -vv -e integrate-test
result:
name: CI
if: always()
+ needs:
+ - dead-link
+ - integrate-test
+ - local-ci
runs-on: ubuntu-latest
steps:
- name: Status
run: |
- echo success
+ if [[ ${{ needs.dead-link.result }} != 'success' ]] || \
+ [[ ${{ needs.integrate-test.result }} != 'success' ]] || \
+ [[ ${{ needs.local-ci.result }} != 'success' ]]; then
+ echo "CI Failed!"
+ exit -1
+ fi
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..080eb58
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,15 @@
+# Editor
+.idea/
+.vscode/
+
+# Cache
+__pycache__/
+.tox/
+
+# Build
+build/
+*egg-info/
+
+# Test coverage
+.coverage
+htmlcov/
diff --git a/.licenserc.yaml b/.licenserc.yaml
new file mode 100644
index 0000000..89bb2bf
--- /dev/null
+++ b/.licenserc.yaml
@@ -0,0 +1,30 @@
+# Licensed to 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. Apache Software Foundation (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.
+
+header:
+ license:
+ spdx-id: Apache-2.0
+ copyright-owner: Apache Software Foundation
+
+ paths-ignore:
+ - NOTICE
+ - LICENSE
+ - 'dist'
+ - '**/*.md'
+ - '**/.gitkeep'
+
+ comment: on-failure
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9e817a5..a767ea5 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -23,13 +23,14 @@ default_language_version:
# force all python hooks to run python3
python: python3
repos:
+ # Python API Hooks
- repo: https://github.com/pycqa/isort
rev: 5.10.1
hooks:
- id: isort
name: isort (python)
- repo: https://github.com/psf/black
- rev: 22.1.0
+ rev: 22.3.0
hooks:
- id: black
- repo: https://github.com/pycqa/flake8
@@ -37,20 +38,20 @@ repos:
hooks:
- id: flake8
additional_dependencies: [
- 'flake8-docstrings>=1.6',
- 'flake8-black>=0.2',
+ 'flake8-docstrings>=1.6',
+ 'flake8-black>=0.2',
]
# pre-commit run in the root, so we have to point out the full path of configuration
args: [
- --config,
- dolphinscheduler-python/pydolphinscheduler/.flake8
+ --config,
+ .flake8
]
- repo: https://github.com/pycqa/autoflake
rev: v1.4
hooks:
- id: autoflake
args: [
- --remove-all-unused-imports,
- --ignore-init-module-imports,
- --in-place
+ --remove-all-unused-imports,
+ --ignore-init-module-imports,
+ --in-place
]
diff --git a/DEVELOP.md b/DEVELOP.md
index efa25a2..2d1a80c 100644
--- a/DEVELOP.md
+++ b/DEVELOP.md
@@ -27,19 +27,17 @@ store or execute it. We here use [py4j][py4j] to dynamically access Java Virtual
**PyDolphinScheduler** use GitHub to hold all source code, you should clone the code before you do same change.
```shell
-git clone git@github.com:apache/dolphinscheduler.git
+git clone git@github.com:apache/dolphinscheduler-sdk-python.git
```
Now, we should install all dependence to make sure we could run test or check code style locally
```shell
-cd dolphinscheduler/dolphinscheduler-python/pydolphinscheduler
python -m pip install -e '.[dev]'
```
Next, we have to open pydolphinscheduler project in you editor. We recommend you use [pycharm][pycharm]
-instead of [IntelliJ IDEA][idea] to open it. And you could just open directory
-`dolphinscheduler-python/pydolphinscheduler` instead of `dolphinscheduler-python`.
+instead of [IntelliJ IDEA][idea] to open it.
## Brief Concept
@@ -208,8 +206,7 @@ this command output.
Integrate Test can not run when you execute command `tox -e local-ci` because it needs external environment
including [Docker](https://docs.docker.com/get-docker/) and specific image build by [maven](https://maven.apache.org/install.html).
-Here we would show you the step to run integrate test in directory `dolphinscheduler-python/pydolphinscheduler/tests/integration`.
-There are two ways to run integrate tests.
+Here we would show you the step to run integrate test in directory `tests/integration`. There are two ways to run integrate tests.
#### Method 1: Launch Docker Container Locally
@@ -249,7 +246,8 @@ When you add a new package in pydolphinscheduler, you should also add the packag
When you change public class, method or interface, you should change the [UPDATING.md](./UPDATING.md) to notice
users who may use it in other way.
-<!-- content -->
+## Reference
+
[py4j]: https://www.py4j.org/index.html
[pycharm]: https://www.jetbrains.com/pycharm
[idea]: https://www.jetbrains.com/idea/
@@ -260,4 +258,5 @@ users who may use it in other way.
[black-editor]: https://black.readthedocs.io/en/stable/integrations/editors.html#pycharm-intellij-idea
[coverage]: https://coverage.readthedocs.io/en/stable/
[isort]: https://pycqa.github.io/isort/index.html
-[sphinx]: https://www.sphinx-doc.org/en/master/
+[sphinx]: https://www.sphinx-doc.org/en/master
+
diff --git a/README.md b/README.md
index 7fc73d6..a311a86 100644
--- a/README.md
+++ b/README.md
@@ -53,7 +53,7 @@ Here we show you how to install and run a simple example of pydolphinscheduler
### Start Server And Run Example
Before you run an example, you have to start backend server. You could follow
-[development setup](../../docs/docs/en/contribute/development-environment-setup.md)
+[development setup](https://dolphinscheduler.apache.org/en-us/docs/dev/user_doc/contribute/development-environment-setup.html)
section "DolphinScheduler Standalone Quick Start" to set up developer environment. You have to start backend
and frontend server in this step, which mean that you could view DolphinScheduler UI in your browser with URL
http://localhost:12345/dolphinscheduler
@@ -65,7 +65,7 @@ And for now we could run a simple example by:
```shell
# Please make sure your terminal could
-curl https://raw.githubusercontent.com/apache/dolphinscheduler/dev/dolphinscheduler-python/pydolphinscheduler/examples/tutorial.py -o ./tutorial.py
+curl https://raw.githubusercontent.com/apache/dolphinscheduler-sdk-python/main/src/pydolphinscheduler/examples/tutorial.py -o ./tutorial.py
python ./tutorial.py
```
diff --git a/RELEASE.md b/RELEASE.md
index e00ef05..b13a0e9 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -22,11 +22,13 @@ under the License.
**PyDolphinScheduler** office release is in [ASF Distribution Directory](https://downloads.apache.org/dolphinscheduler/),
and it should be released together with [apache-dolphinscheduler](https://github.com/apache/dolphinscheduler).
+<!--
## To ASF Distribution Directory
You could release to [ASF Distribution Directory](https://downloads.apache.org/dolphinscheduler/) according to
[release guide](../../docs/docs/en/contribute/release/release-prepare.md) in DolphinScheduler
website.
+-->
## To PyPi
diff --git a/UPDATING.md b/UPDATING.md
index e918b1e..b298c3b 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -25,6 +25,7 @@ It started after version 2.0.5 released
## dev
* Remove parameter ``task_location`` in process definition and Java Gateway service ([#11681](https://github.com/apache/dolphinscheduler/pull/11681))
+* Remove the spark version of spark task ([#11860](https://github.com/apache/dolphinscheduler/pull/11860)).
## 3.0.0
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 23fc117..c7aa6d3 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -88,7 +88,7 @@ html_sidebars = {
}
# Match all exists tag for pydolphinscheduler expect version 2.0.4(not release apache dolphinscheduler)
smv_tag_whitelist = r"^(?!2.0.4)\d+\.\d+\.\d+$"
-smv_branch_whitelist = "dev"
+smv_branch_whitelist = "main"
smv_remote_whitelist = r"^(origin|upstream)$"
smv_released_pattern = "^refs/tags/.*$"
smv_outputdir_format = "versions/{ref.name}"
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 24ad107..4dc0a94 100644
--- a/docs/source/index.rst
+++ b/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
==================
diff --git a/docs/source/resources_plugin/develop.rst b/docs/source/resources_plugin/develop.rst
new file mode 100644
index 0000000..e7d90ea
--- /dev/null
+++ b/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 plugin 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 plugin.
+
+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]
diff --git a/docs/source/index.rst b/docs/source/resources_plugin/github.rst
similarity index 52%
copy from docs/source/index.rst
copy to docs/source/resources_plugin/github.rst
index 24ad107..b302337 100644
--- a/docs/source/index.rst
+++ b/docs/source/resources_plugin/github.rst
@@ -15,31 +15,21 @@
specific language governing permissions and limitations
under the License.
-PyDolphinScheduler
-==================
+GitHub
+======
-**PyDolphinScheduler** is Python API for `Apache DolphinScheduler <https://dolphinscheduler.apache.org>`_,
-which allow you definition your workflow by Python code, aka workflow-as-codes.
+`GitHub` is a github resource plugin for pydolphinscheduler.
-I could go and find how to :ref:`install <start:getting started>` the project. Or if you want to see simply example
-then go and see :doc:`tutorial` for more detail.
+When using a github resource plugin, you only need to add the `resource_plugin` parameter in the task subclass or workflow definition,
+such as `resource_plugin=GitHub(prefix="https://github.com/xxx", access_token="ghpxx")`.
+The token parameter is optional. You need to add it when your repository is a private repository.
+You can view this `document <https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/creating-a-personal-access-token>`_
+when creating a token.
-.. toctree::
- :maxdepth: 2
+For the specific use of resource plugins, you can see `How to use` in :doc:`resource-plugin`
- start
- tutorial
- concept
- tasks/index
- howto/index
- cli
- config
- api
+Dive Into
+---------
-Indices and tables
-==================
-
-* :ref:`genindex`
-* :ref:`modindex`
-* :ref:`search`
+.. automodule:: pydolphinscheduler.resources_plugin.github
\ No newline at end of file
diff --git a/docs/source/resources_plugin/gitlab.rst b/docs/source/resources_plugin/gitlab.rst
new file mode 100644
index 0000000..fdf43c9
--- /dev/null
+++ b/docs/source/resources_plugin/gitlab.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.
+
+GitLab
+======
+
+`GitLab` is a gitlab resource plugin for pydolphinscheduler.
+
+When using a gitlab resource plugin, you only need to add the `resource_plugin` parameter in the task subclass or workflow definition,
+such as `resource_plugin=GitLab(prefix="xxx")`, if it is a public repository.
+
+If it is a private or Internal repository, you can use three ways to obtain authentication.
+
+The first is `Personal Access Tokens`, using `resource_plugin=GitLab(prefix="xxx", private_token="xxx")`.
+
+The second method is to obtain authentication through `username` and `password`:
+
+using `resource_plugin=GitLab(prefix="xxx", username="username", password="pwd")`.
+
+The third method is to obtain authentication through `OAuth Token`:
+
+using `resource_plugin=GitLab(prefix="xxx", oauth_token="xx")`.
+
+You can view this `document <https://docs.gitlab.com/ee/user/profile/personal_access_tokens.html#create-a-personal-access-token>`_
+when creating a `Personal Access Tokens`.
+
+For the specific use of resource plugins, you can see `How to use` in :doc:`resource-plugin`
+
+Dive Into
+---------
+
+.. automodule:: pydolphinscheduler.resources_plugin.gitlab
\ No newline at end of file
diff --git a/docs/source/index.rst b/docs/source/resources_plugin/index.rst
similarity index 58%
copy from docs/source/index.rst
copy to docs/source/resources_plugin/index.rst
index 24ad107..c984f06 100644
--- a/docs/source/index.rst
+++ b/docs/source/resources_plugin/index.rst
@@ -15,31 +15,18 @@
specific language governing permissions and limitations
under the License.
-PyDolphinScheduler
-==================
-
-**PyDolphinScheduler** is Python API for `Apache DolphinScheduler <https://dolphinscheduler.apache.org>`_,
-which allow you definition your workflow by Python code, aka workflow-as-codes.
-
-I could go and find how to :ref:`install <start:getting started>` the project. Or if you want to see simply example
-then go and see :doc:`tutorial` for more detail.
+Resources_plugin
+================
+In this section
.. toctree::
- :maxdepth: 2
-
- start
- tutorial
- concept
- tasks/index
- howto/index
- cli
- config
- api
-
-Indices and tables
-==================
-
-* :ref:`genindex`
-* :ref:`modindex`
-* :ref:`search`
+ :maxdepth: 1
+
+ develop
+ resource-plugin
+ local
+ github
+ gitlab
+ oss
+ s3
\ No newline at end of file
diff --git a/docs/source/index.rst b/docs/source/resources_plugin/local.rst
similarity index 57%
copy from docs/source/index.rst
copy to docs/source/resources_plugin/local.rst
index 24ad107..5da025a 100644
--- a/docs/source/index.rst
+++ b/docs/source/resources_plugin/local.rst
@@ -15,31 +15,18 @@
specific language governing permissions and limitations
under the License.
-PyDolphinScheduler
-==================
+Local
+=====
-**PyDolphinScheduler** is Python API for `Apache DolphinScheduler <https://dolphinscheduler.apache.org>`_,
-which allow you definition your workflow by Python code, aka workflow-as-codes.
+`Local` is a local resource plugin for pydolphinscheduler.
-I could go and find how to :ref:`install <start:getting started>` the project. Or if you want to see simply example
-then go and see :doc:`tutorial` for more detail.
+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")`.
-.. toctree::
- :maxdepth: 2
+For the specific use of resource plugins, you can see `How to use` in :doc:`./resource-plugin`
- start
- tutorial
- concept
- tasks/index
- howto/index
- cli
- config
- api
+Dive Into
+---------
-Indices and tables
-==================
-
-* :ref:`genindex`
-* :ref:`modindex`
-* :ref:`search`
+.. automodule:: pydolphinscheduler.resources_plugin.local
\ No newline at end of file
diff --git a/docs/source/resources_plugin/oss.rst b/docs/source/resources_plugin/oss.rst
new file mode 100644
index 0000000..fbb6785
--- /dev/null
+++ b/docs/source/resources_plugin/oss.rst
@@ -0,0 +1,44 @@
+.. 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.
+
+OSS
+===
+
+`OSS` is a Aliyun OSS resource plugin for pydolphinscheduler.
+
+When using a OSS resource plugin, you only need to add the `resource_plugin` parameter in the task subclass or workflow definition,
+such as `resource_plugin=OSS(prefix="xxx")`, if the file is publicly readable.
+
+When the file is private, using `resource_plugin=OSS(prefix="xxx", access_key_id="xxx", access_key_secret="xxx")`
+
+Notice
+The read permission of files in a bucket is inherited from the bucket by default. In other words, if the bucket is private,
+the files in it are also private.
+
+But the read permission of the files in the bucket can be changed, in other words, the files in the private bucket can also be read publicly.
+
+So whether the `AccessKey` is needed depends on whether the file is private or not.
+
+You can view this `document <https://www.alibabacloud.com/help/en/tablestore/latest/how-can-i-obtain-an-accesskey-pair>`_
+when creating a pair `AccessKey`.
+
+For the specific use of resource plugins, you can see `How to use` in :doc:`resource-plugin`
+
+Dive Into
+---------
+
+.. automodule:: pydolphinscheduler.resources_plugin.OSS
diff --git a/docs/source/resources_plugin/resource-plugin.rst b/docs/source/resources_plugin/resource-plugin.rst
new file mode 100644
index 0000000..2a32526
--- /dev/null
+++ b/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 plugin can be used in task subclasses and workflows. You can use the resource plugin by adding the `resource_plugin` parameter when they are initialized.
+For example, local resource plugin, add `resource_plugin = Local("/tmp")`.
+
+The resource plugin we currently support are `local`, `github`, `gitlab`, `OSS`, `S3`.
+
+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.
\ No newline at end of file
diff --git a/docs/source/index.rst b/docs/source/resources_plugin/s3.rst
similarity index 53%
copy from docs/source/index.rst
copy to docs/source/resources_plugin/s3.rst
index 24ad107..f5bc1d3 100644
--- a/docs/source/index.rst
+++ b/docs/source/resources_plugin/s3.rst
@@ -15,31 +15,22 @@
specific language governing permissions and limitations
under the License.
-PyDolphinScheduler
-==================
+S3
+==
-**PyDolphinScheduler** is Python API for `Apache DolphinScheduler <https://dolphinscheduler.apache.org>`_,
-which allow you definition your workflow by Python code, aka workflow-as-codes.
+`S3` is a Amazon S3 resource plugin for pydolphinscheduler.
-I could go and find how to :ref:`install <start:getting started>` the project. Or if you want to see simply example
-then go and see :doc:`tutorial` for more detail.
+When using a Amazon S3 resource plugin, you only need to add the `resource_plugin` parameter in the task subclass or workflow definition,
+such as `resource_plugin=S3(prefix="xxx")`, if the file is publicly readable.
+When the file is private, using `resource_plugin=S3(prefix="xxx", access_key_id="xxx", access_key_secret="xxx")`
-.. toctree::
- :maxdepth: 2
+You can view this `document <https://docs.aws.amazon.com/general/latest/gr/aws-access-keys-best-practices.html>`_
+when creating a pair `AccessKey`.
- start
- tutorial
- concept
- tasks/index
- howto/index
- cli
- config
- api
+For the specific use of resource plugins, you can see `How to use` in :doc:`resource-plugin`
-Indices and tables
-==================
+Dive Into
+---------
-* :ref:`genindex`
-* :ref:`modindex`
-* :ref:`search`
+.. automodule:: pydolphinscheduler.resources_plugin.S3
diff --git a/docs/source/start.rst b/docs/source/start.rst
index 270b5b8..aa86f71 100644
--- a/docs/source/start.rst
+++ b/docs/source/start.rst
@@ -79,9 +79,9 @@ which we hold in GitHub
.. code-block:: bash
# Clone Apache DolphinScheduler repository
- git clone git@github.com:apache/dolphinscheduler.git
+ git clone git@github.com:apache/dolphinscheduler-sdk-python.git
# Install PyDolphinScheduler in develop mode
- cd dolphinscheduler-python/pydolphinscheduler && python -m pip install -e .
+ python -m pip install -e .
After you installed *PyDolphinScheduler*, please remember `start Python Gateway Service`_
which waiting for *PyDolphinScheduler*'s workflow definition require.
@@ -92,7 +92,7 @@ package directly and do not care about other code(including Python gateway servi
.. code-block:: bash
# Must escape the '&' character by adding '\'
- pip install -e "git+https://github.com/apache/dolphinscheduler.git#egg=apache-dolphinscheduler&subdirectory=dolphinscheduler-python/pydolphinscheduler"
+ pip install -e "git+https://github.com/apache/dolphinscheduler-sdk-python.git#egg=apache-dolphinscheduler"
Start Python Gateway Service
----------------------------
@@ -132,7 +132,7 @@ single bash command to get it
.. code-block:: bash
- wget https://raw.githubusercontent.com/apache/dolphinscheduler/dev/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/examples/tutorial.py
+ wget https://raw.githubusercontent.com/apache/dolphinscheduler-sdk-python/main/src/pydolphinscheduler/examples/tutorial.py
or you could copy-paste the content from `tutorial source code`_. And then you could run the example in your
terminal
@@ -168,4 +168,4 @@ if you already know the basic usage or concept of *PyDolphinScheduler*, you coul
.. _`instructions for all platforms here`: https://wiki.python.org/moin/BeginnersGuide/Download
.. _`Apache DolphinScheduler`: https://dolphinscheduler.apache.org
.. _`install Apache DolphinScheduler`: https://dolphinscheduler.apache.org/en-us/docs/latest/user_doc/guide/installation/standalone.html
-.. _`tutorial source code`: https://raw.githubusercontent.com/apache/dolphinscheduler/dev/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/examples/tutorial.py
+.. _`tutorial source code`: https://raw.githubusercontent.com/apache/dolphinscheduler-sdk-python/main/src/pydolphinscheduler/examples/tutorial.py
diff --git a/docs/source/tutorial.rst b/docs/source/tutorial.rst
index 57d21b2..16e0d35 100644
--- a/docs/source/tutorial.rst
+++ b/docs/source/tutorial.rst
@@ -38,7 +38,7 @@ There are two types of tutorials: traditional and task decorator.
supported. But it is helpful if your workflow is all built with Python or if you already have some Python
workflow code and want to migrate them to pydolphinscheduler.
- **YAML File**: We can use pydolphinscheduler CLI to create process using YAML file: :code:`pydolphinscheduler yaml -f tutorial.yaml`.
- We can find more YAML file examples in `examples/yaml_define <https://github.com/apache/dolphinscheduler/tree/dev/dolphinscheduler-python/pydolphinscheduler/examples/yaml_define>`_
+ We can find more YAML file examples in `examples/yaml_define <https://github.com/apache/dolphinscheduler-sdk-python/tree/main/examples/yaml_define>`_
.. tab:: Tradition
diff --git a/examples/yaml_define/Spark.yaml b/examples/yaml_define/Spark.yaml
index 6132b8d..e45514b 100644
--- a/examples/yaml_define/Spark.yaml
+++ b/examples/yaml_define/Spark.yaml
@@ -27,4 +27,3 @@ tasks:
main_package: test_java.jar
program_type: SCALA
deploy_mode: local
- spark_version: SPARK1
diff --git a/setup.py b/setup.py
index 355b3a5..bf71d7d 100644
--- a/setup.py
+++ b/setup.py
@@ -32,10 +32,13 @@ if sys.version_info[0] < 3:
logger = logging.getLogger(__name__)
-version = "3.1.0"
+version = "dev"
# Start package required
prod = [
+ "boto3>=1.23.10",
+ "oss2>=2.16.0",
+ "python-gitlab>=2.10.1",
"click>=8.0.0",
"py4j~=0.10",
"ruamel.yaml",
@@ -140,11 +143,8 @@ setup(
project_urls={
"Homepage": "https://dolphinscheduler.apache.org",
"Documentation": "https://dolphinscheduler.apache.org/python/dev/index.html",
- "Source": "https://github.com/apache/dolphinscheduler/tree/dev/dolphinscheduler-python/"
- "pydolphinscheduler",
- "Issue Tracker": "https://github.com/apache/dolphinscheduler/issues?"
- "q=is%3Aissue+is%3Aopen+label%3APython",
- "Discussion": "https://github.com/apache/dolphinscheduler/discussions",
+ "Source": "https://github.com/apache/dolphinscheduler-sdk-python",
+ "Issue Tracker": "https://github.com/apache/dolphinscheduler-sdk-python/issues",
"Twitter": "https://twitter.com/dolphinschedule",
},
packages=find_packages(where="src"),
diff --git a/src/pydolphinscheduler/constants.py b/src/pydolphinscheduler/constants.py
index fd640c5..bedbbf2 100644
--- a/src/pydolphinscheduler/constants.py
+++ b/src/pydolphinscheduler/constants.py
@@ -111,3 +111,12 @@ class ResourceKey(str):
"""Constants for key of resource."""
ID = "id"
+
+
+class Symbol(str):
+ """Constants for symbol."""
+
+ SLASH = "/"
+ POINT = "."
+ COMMA = ","
+ UNDERLINE = "_"
diff --git a/src/pydolphinscheduler/core/base.py b/src/pydolphinscheduler/core/base.py
deleted file mode 100644
index 690351a..0000000
--- a/src/pydolphinscheduler/core/base.py
+++ /dev/null
@@ -1,74 +0,0 @@
-# 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 Base object."""
-
-from typing import Dict, Optional
-
-# from pydolphinscheduler.side.user import User
-from pydolphinscheduler.utils.string import attr2camel
-
-
-class Base:
- """DolphinScheduler Base object."""
-
- # Object key attribute, to test whether object equals and so on.
- _KEY_ATTR: set = {"name", "description"}
-
- # Object defines attribute, use when needs to communicate with Java gateway server.
- _DEFINE_ATTR: set = set()
-
- # Object default attribute, will add those attribute to `_DEFINE_ATTR` when init assign missing.
- _DEFAULT_ATTR: Dict = {}
-
- def __init__(self, name: str, description: Optional[str] = None):
- self.name = name
- self.description = description
-
- def __repr__(self) -> str:
- return f'<{type(self).__name__}: name="{self.name}">'
-
- def __eq__(self, other):
- return type(self) == type(other) and all(
- getattr(self, a, None) == getattr(other, a, None) for a in self._KEY_ATTR
- )
-
- def get_define_custom(
- self, camel_attr: bool = True, custom_attr: set = None
- ) -> Dict:
- """Get object definition attribute by given attr set."""
- content = {}
- for attr in custom_attr:
- val = getattr(self, attr, None)
- if camel_attr:
- content[attr2camel(attr)] = val
- else:
- content[attr] = val
- return content
-
- def get_define(self, camel_attr: bool = True) -> Dict:
- """Get object definition attribute communicate to Java gateway server.
-
- use attribute `self._DEFINE_ATTR` to determine which attributes should including when
- object tries to communicate with Java gateway server.
- """
- content = self.get_define_custom(camel_attr, self._DEFINE_ATTR)
- update_default = {
- k: self._DEFAULT_ATTR.get(k) for k in self._DEFAULT_ATTR if k not in content
- }
- content.update(update_default)
- return content
diff --git a/src/pydolphinscheduler/core/base_side.py b/src/pydolphinscheduler/core/base_side.py
deleted file mode 100644
index dca1f12..0000000
--- a/src/pydolphinscheduler/core/base_side.py
+++ /dev/null
@@ -1,40 +0,0 @@
-# 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.
-
-"""Module for side object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core import configuration
-from pydolphinscheduler.core.base import Base
-
-
-class BaseSide(Base):
- """Base class for side object, it declare base behavior for them."""
-
- def __init__(self, name: str, description: Optional[str] = None):
- super().__init__(name, description)
-
- @classmethod
- def create_if_not_exists(
- cls,
- # TODO comment for avoiding cycle import
- # user: Optional[User] = ProcessDefinitionDefault.USER
- user=configuration.WORKFLOW_USER,
- ):
- """Create Base if not exists."""
- raise NotImplementedError
diff --git a/src/pydolphinscheduler/core/configuration.py b/src/pydolphinscheduler/core/configuration.py
deleted file mode 100644
index 860f986..0000000
--- a/src/pydolphinscheduler/core/configuration.py
+++ /dev/null
@@ -1,193 +0,0 @@
-# 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.
-
-"""Configuration module for pydolphinscheduler."""
-import os
-from pathlib import Path
-from typing import Any
-
-from pydolphinscheduler.exceptions import PyDSConfException
-from pydolphinscheduler.utils import file
-from pydolphinscheduler.utils.yaml_parser import YamlParser
-
-BUILD_IN_CONFIG_PATH = Path(__file__).resolve().parent.joinpath("default_config.yaml")
-
-
-def config_path() -> Path:
- """Get the path of pydolphinscheduler configuration file."""
- pyds_home = os.environ.get("PYDS_HOME", "~/pydolphinscheduler")
- config_file_path = Path(pyds_home).joinpath("config.yaml").expanduser()
- return config_file_path
-
-
-def get_configs() -> YamlParser:
- """Get all configuration settings from configuration file.
-
- Will use custom configuration file first if it exists, otherwise default configuration file in
- default path.
- """
- path = str(config_path()) if config_path().exists() else BUILD_IN_CONFIG_PATH
- with open(path, mode="r") as f:
- return YamlParser(f.read())
-
-
-def init_config_file() -> None:
- """Initialize configuration file by default configs."""
- if config_path().exists():
- raise PyDSConfException(
- "Initialize configuration false to avoid overwrite configure by accident, file already exists "
- "in %s, if you wan to overwrite the exists configure please remove the exists file manually.",
- str(config_path()),
- )
- file.write(content=str(get_configs()), to_path=str(config_path()))
-
-
-def get_single_config(key: str) -> Any:
- """Get single config to configuration file.
-
- Support get from nested keys by delimiter ``.``.
-
- For example, yaml config as below:
-
- .. code-block:: yaml
-
- one:
- two1:
- three: value1
- two2: value2
-
- you could get ``value1`` and ``value2`` by nested path
-
- .. code-block:: python
-
- value1 = get_single_config("one.two1.three")
- value2 = get_single_config("one.two2")
-
- :param key: The config key want to get it value.
- """
- config = get_configs()
- if key not in config:
- raise PyDSConfException(
- "Configuration path %s do not exists. Can not get configuration.", key
- )
- return config[key]
-
-
-def set_single_config(key: str, value: Any) -> None:
- """Change single config to configuration file.
-
- For example, yaml config as below:
-
- .. code-block:: yaml
-
- one:
- two1:
- three: value1
- two2: value2
-
- you could change ``value1`` to ``value3``, also change ``value2`` to ``value4`` by nested path assigned
-
- .. code-block:: python
-
- set_single_config["one.two1.three"] = "value3"
- set_single_config["one.two2"] = "value4"
-
- :param key: The config key want change.
- :param value: The new value want to set.
- """
- config = get_configs()
- if key not in config:
- raise PyDSConfException(
- "Configuration path %s do not exists. Can not set configuration.", key
- )
- config[key] = value
- file.write(content=str(config), to_path=str(config_path()), overwrite=True)
-
-
-def get_int(val: Any) -> int:
- """Covert value to int."""
- return int(val)
-
-
-def get_bool(val: Any) -> bool:
- """Covert value to boolean."""
- if isinstance(val, str):
- return val.lower() in {"true", "t"}
- elif isinstance(val, int):
- return val == 1
- else:
- return bool(val)
-
-
-# Start Common Configuration Settings
-
-# Add configs as module variables to avoid read configuration multiple times when
-# Get common configuration setting
-# Set or get multiple configs in single time
-configs: YamlParser = get_configs()
-
-# Java Gateway Settings
-JAVA_GATEWAY_ADDRESS = os.environ.get(
- "PYDS_JAVA_GATEWAY_ADDRESS", configs.get("java_gateway.address")
-)
-JAVA_GATEWAY_PORT = get_int(
- os.environ.get("PYDS_JAVA_GATEWAY_PORT", configs.get("java_gateway.port"))
-)
-JAVA_GATEWAY_AUTO_CONVERT = get_bool(
- os.environ.get(
- "PYDS_JAVA_GATEWAY_AUTO_CONVERT", configs.get("java_gateway.auto_convert")
- )
-)
-
-# User Settings
-USER_NAME = os.environ.get("PYDS_USER_NAME", configs.get("default.user.name"))
-USER_PASSWORD = os.environ.get(
- "PYDS_USER_PASSWORD", configs.get("default.user.password")
-)
-USER_EMAIL = os.environ.get("PYDS_USER_EMAIL", configs.get("default.user.email"))
-USER_PHONE = str(os.environ.get("PYDS_USER_PHONE", configs.get("default.user.phone")))
-USER_STATE = get_int(
- os.environ.get("PYDS_USER_STATE", configs.get("default.user.state"))
-)
-
-# Workflow Settings
-WORKFLOW_PROJECT = os.environ.get(
- "PYDS_WORKFLOW_PROJECT", configs.get("default.workflow.project")
-)
-WORKFLOW_TENANT = os.environ.get(
- "PYDS_WORKFLOW_TENANT", configs.get("default.workflow.tenant")
-)
-WORKFLOW_USER = os.environ.get(
- "PYDS_WORKFLOW_USER", configs.get("default.workflow.user")
-)
-WORKFLOW_QUEUE = os.environ.get(
- "PYDS_WORKFLOW_QUEUE", configs.get("default.workflow.queue")
-)
-WORKFLOW_RELEASE_STATE = os.environ.get(
- "PYDS_WORKFLOW_RELEASE_STATE", configs.get("default.workflow.release_state")
-)
-WORKFLOW_WORKER_GROUP = os.environ.get(
- "PYDS_WORKFLOW_WORKER_GROUP", configs.get("default.workflow.worker_group")
-)
-WORKFLOW_TIME_ZONE = os.environ.get(
- "PYDS_WORKFLOW_TIME_ZONE", configs.get("default.workflow.time_zone")
-)
-WORKFLOW_WARNING_TYPE = os.environ.get(
- "PYDS_WORKFLOW_WARNING_TYPE", configs.get("default.workflow.warning_type")
-)
-
-# End Common Configuration Setting
diff --git a/src/pydolphinscheduler/core/process_definition.py b/src/pydolphinscheduler/core/process_definition.py
index df05b01..62de7ed 100644
--- a/src/pydolphinscheduler/core/process_definition.py
+++ b/src/pydolphinscheduler/core/process_definition.py
@@ -24,6 +24,7 @@ from typing import Any, Dict, List, Optional, Set
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 @@ class ProcessDefinition(Base):
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 @@ class ProcessDefinition(Base):
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
diff --git a/src/pydolphinscheduler/core/resource_plugin.py b/src/pydolphinscheduler/core/resource_plugin.py
new file mode 100644
index 0000000..8b500d1
--- /dev/null
+++ b/src/pydolphinscheduler/core/resource_plugin.py
@@ -0,0 +1,58 @@
+# 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
+
+from pydolphinscheduler.exceptions import PyResPluginException
+
+
+# [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]
+
+ def get_index(self, s: str, x, n):
+ """Find the subscript of the nth occurrence of the X character in the string s."""
+ if n <= s.count(x):
+ all_index = [key for key, value in enumerate(s) if value == x]
+ return all_index[n - 1]
+ else:
+ raise PyResPluginException("Incomplete path.")
+
+
+# [end resource_plugin_definition]
diff --git a/src/pydolphinscheduler/core/task.py b/src/pydolphinscheduler/core/task.py
index 5cbd21d..3fec31f 100644
--- a/src/pydolphinscheduler/core/task.py
+++ b/src/pydolphinscheduler/core/task.py
@@ -17,6 +17,7 @@
"""DolphinScheduler Task and TaskRelation object."""
import copy
+import types
from logging import getLogger
from typing import Dict, List, Optional, Sequence, Set, Tuple, Union
@@ -24,6 +25,7 @@ from pydolphinscheduler import configuration
from pydolphinscheduler.constants import (
Delimiter,
ResourceKey,
+ Symbol,
TaskFlag,
TaskPriority,
TaskTimeoutFlag,
@@ -33,7 +35,8 @@ from pydolphinscheduler.core.process_definition import (
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
@@ -112,6 +115,9 @@ class Task(Base):
# task custom attribute define in sub class and will append to `task_params` property
_task_custom_attr: set = set()
+ ext: set = None
+ ext_attr: Union[str, types.FunctionType] = None
+
DEFAULT_CONDITION_RESULT = {"successNode": [""], "failedNode": [""]}
def __init__(
@@ -135,6 +141,7 @@ class Task(Base):
dependence: Optional[Dict] = None,
wait_start_timeout: Optional[Dict] = None,
condition_result: Optional[Dict] = None,
+ resource_plugin: Optional[ResourcePlugin] = None,
):
super().__init__(name, description)
@@ -177,6 +184,8 @@ class Task(Base):
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]:
@@ -244,6 +253,47 @@ class Task(Base):
custom_attr = self._get_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 isinstance(_ext_attr, str) and _ext_attr.endswith(tuple(self.ext)):
+ res = self.get_plugin()
+ content = res.read_file(_ext_attr)
+ setattr(self, self.ext_attr.lstrip(Symbol.UNDERLINE), content)
+ else:
+ if self.resource_plugin is not None or (
+ self.process_definition is not None
+ and self.process_definition.resource_plugin is not None
+ ):
+ index = _ext_attr.rfind(Symbol.POINT)
+ if index != -1:
+ raise ValueError(
+ "This task does not support files with suffix {}, only supports {}".format(
+ _ext_attr[index:],
+ Symbol.COMMA.join(str(suf) for suf in self.ext),
+ )
+ )
+ setattr(self, self.ext_attr.lstrip(Symbol.UNDERLINE), _ext_attr)
+
def __hash__(self):
return hash(self.code)
diff --git a/src/pydolphinscheduler/examples/tutorial_resource_plugin.py b/src/pydolphinscheduler/examples/tutorial_resource_plugin.py
new file mode 100644
index 0000000..5b02022
--- /dev/null
+++ b/src/pydolphinscheduler/examples/tutorial_resource_plugin.py
@@ -0,0 +1,64 @@
+# 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.
+
+r"""
+A tutorial example take you to experience pydolphinscheduler resource plugin.
+
+Resource plug-ins can be defined in workflows and tasks
+
+it will instantiate and run all the task it have.
+"""
+import os
+from pathlib import Path
+
+# [start tutorial_resource_plugin]
+# [start package_import]
+# Import ProcessDefinition object to define your workflow attributes
+from pydolphinscheduler.core.process_definition import ProcessDefinition
+
+# Import task Shell object cause we would create some shell tasks later
+from pydolphinscheduler.resources_plugin.local import Local
+from pydolphinscheduler.tasks.shell import Shell
+
+# [end package_import]
+
+# [start workflow_declare]
+with ProcessDefinition(
+ name="tutorial_resource_plugin",
+ schedule="0 0 0 * * ? *",
+ start_time="2021-01-01",
+ tenant="tenant_exists",
+ resource_plugin=Local("/tmp"),
+) as process_definition:
+ # [end workflow_declare]
+ # [start task_declare]
+ file = "resource.sh"
+ path = Path("/tmp").joinpath(file)
+ with open(str(path), "w") as f:
+ f.write("echo tutorial resource plugin")
+ task_parent = Shell(
+ name="local-resource-example",
+ command=file,
+ )
+ print(task_parent.task_params)
+ os.remove(path)
+ # [end task_declare]
+
+ # [start submit_or_run]
+ process_definition.run()
+ # [end submit_or_run]
+# [end tutorial_resource_plugin]
diff --git a/src/pydolphinscheduler/exceptions.py b/src/pydolphinscheduler/exceptions.py
index 4d70a58..5b0d1bb 100644
--- a/src/pydolphinscheduler/exceptions.py
+++ b/src/pydolphinscheduler/exceptions.py
@@ -40,3 +40,7 @@ class PyDSProcessDefinitionNotAssignException(PyDSBaseException):
class PyDSConfException(PyDSBaseException):
"""Exception for pydolphinscheduler configuration error."""
+
+
+class PyResPluginException(PyDSBaseException):
+ """Exception for pydolphinscheduler resource plugin error."""
diff --git a/src/pydolphinscheduler/java_gateway.py b/src/pydolphinscheduler/java_gateway.py
index 0ff74ba..54bb0a3 100644
--- a/src/pydolphinscheduler/java_gateway.py
+++ b/src/pydolphinscheduler/java_gateway.py
@@ -17,15 +17,20 @@
"""Module java gateway, contain gateway behavior."""
+import contextlib
+from logging import getLogger
from typing import Any, Optional
from py4j.java_collections import JavaMap
from py4j.java_gateway import GatewayParameters, JavaGateway
+from py4j.protocol import Py4JError
-from pydolphinscheduler import configuration
+from pydolphinscheduler import __version__, configuration
from pydolphinscheduler.constants import JavaGatewayDefault
from pydolphinscheduler.exceptions import PyDSJavaGatewayException
+logger = getLogger(__name__)
+
def launch_gateway(
address: Optional[str] = None,
@@ -75,6 +80,22 @@ class JavaGate:
auto_convert: Optional[bool] = True,
):
self.java_gateway = launch_gateway(address, port, auto_convert)
+ gateway_version = "unknown"
+ with contextlib.suppress(Py4JError):
+ # 1. Java gateway version is too old: doesn't have method 'getGatewayVersion()'
+ # 2. Error connecting to Java gateway
+ gateway_version = self.get_gateway_version()
+ if gateway_version != __version__:
+ logger.warning(
+ f"Using unmatched version of pydolphinscheduler (version {__version__}) "
+ f"and Java gateway (version {gateway_version}) may cause errors. "
+ "We strongly recommend you to find the matched version "
+ "(check: https://pypi.org/project/apache-dolphinscheduler)"
+ )
+
+ def get_gateway_version(self):
+ """Get the java gateway version, expected to be equal with pydolphinscheduler."""
+ return self.java_gateway.entry_point.getGatewayVersion()
def get_datasource_info(self, name: str):
"""Get datasource info through java gateway."""
@@ -118,6 +139,22 @@ class JavaGate:
user, name, description
)
+ def query_project_by_name(self, user: str, name: str):
+ """Query project through java gateway."""
+ return self.java_gateway.entry_point.queryProjectByName(user, name)
+
+ def update_project(
+ self, user: str, project_code: int, project_name: str, description: str
+ ):
+ """Update project through java gateway."""
+ return self.java_gateway.entry_point.updateProject(
+ user, project_code, project_name, description
+ )
+
+ def delete_project(self, user: str, code: int):
+ """Delete project through java gateway."""
+ return self.java_gateway.entry_point.deleteProject(user, code)
+
def create_tenant(
self, tenant_name: str, queue_name: str, description: Optional[str] = None
):
@@ -126,6 +163,31 @@ class JavaGate:
tenant_name, description, queue_name
)
+ def query_tenant(self, tenant_code: str):
+ """Query tenant through java gateway."""
+ return self.java_gateway.entry_point.queryTenantByCode(tenant_code)
+
+ def grant_tenant_to_user(self, user_name: str, tenant_code: str):
+ """Grant tenant to user through java gateway."""
+ return self.java_gateway.entry_point.grantTenantToUser(user_name, tenant_code)
+
+ def update_tenant(
+ self,
+ user: str,
+ tenant_id: int,
+ code: str,
+ queue_id: int,
+ description: Optional[str] = None,
+ ):
+ """Update tenant through java gateway."""
+ return self.java_gateway.entry_point.updateTenant(
+ user, tenant_id, code, queue_id, description
+ )
+
+ def delete_tenant(self, user: str, tenant_id: int):
+ """Delete tenant through java gateway."""
+ return self.java_gateway.entry_point.deleteTenantById(user, tenant_id)
+
def create_user(
self,
name: str,
@@ -141,6 +203,29 @@ class JavaGate:
name, password, email, phone, tenant, queue, status
)
+ def query_user(self, user_id: int):
+ """Query user through java gateway."""
+ return self.java_gateway.queryUser(user_id)
+
+ def update_user(
+ self,
+ name: str,
+ password: str,
+ email: str,
+ phone: str,
+ tenant: str,
+ queue: str,
+ status: int,
+ ):
+ """Update user through java gateway."""
+ return self.java_gateway.entry_point.updateUser(
+ name, password, email, phone, tenant, queue, status
+ )
+
+ def delete_user(self, name: str, user_id: int):
+ """Delete user through java gateway."""
+ return self.java_gateway.entry_point.deleteUser(name, user_id)
+
def get_dependent_info(
self,
project_name: str,
diff --git a/src/pydolphinscheduler/models/base_side.py b/src/pydolphinscheduler/models/base_side.py
index 67ac88d..99b4007 100644
--- a/src/pydolphinscheduler/models/base_side.py
+++ b/src/pydolphinscheduler/models/base_side.py
@@ -38,3 +38,11 @@ class BaseSide(Base):
):
"""Create Base if not exists."""
raise NotImplementedError
+
+ def delete_all(self):
+ """Delete all method."""
+ if not self:
+ return
+ list_pro = [key for key in self.__dict__.keys()]
+ for key in list_pro:
+ self.__delattr__(key)
diff --git a/src/pydolphinscheduler/models/project.py b/src/pydolphinscheduler/models/project.py
index bebdafd..678332b 100644
--- a/src/pydolphinscheduler/models/project.py
+++ b/src/pydolphinscheduler/models/project.py
@@ -31,11 +31,42 @@ class Project(BaseSide):
self,
name: str = configuration.WORKFLOW_PROJECT,
description: Optional[str] = None,
+ code: Optional[int] = None,
):
super().__init__(name, description)
+ self.code = code
def create_if_not_exists(self, user=configuration.USER_NAME) -> None:
"""Create Project if not exists."""
JavaGate().create_or_grant_project(user, self.name, self.description)
# TODO recover result checker
# gateway_result_checker(result, None)
+
+ @classmethod
+ def get_project_by_name(cls, user=configuration.USER_NAME, name=None) -> "Project":
+ """Get Project by name."""
+ project = JavaGate().query_project_by_name(user, name)
+ if project is None:
+ return cls()
+ return cls(
+ name=project.getName(),
+ description=project.getDescription(),
+ code=project.getCode(),
+ )
+
+ def update(
+ self,
+ user=configuration.USER_NAME,
+ project_code=None,
+ project_name=None,
+ description=None,
+ ) -> None:
+ """Update Project."""
+ JavaGate().update_project(user, project_code, project_name, description)
+ self.name = project_name
+ self.description = description
+
+ def delete(self, user=configuration.USER_NAME) -> None:
+ """Delete Project."""
+ JavaGate().delete_project(user, self.code)
+ self.delete_all()
diff --git a/src/pydolphinscheduler/models/tenant.py b/src/pydolphinscheduler/models/tenant.py
index 6641d9a..09b00cc 100644
--- a/src/pydolphinscheduler/models/tenant.py
+++ b/src/pydolphinscheduler/models/tenant.py
@@ -32,13 +32,49 @@ class Tenant(BaseSide):
name: str = configuration.WORKFLOW_TENANT,
queue: str = configuration.WORKFLOW_QUEUE,
description: Optional[str] = None,
+ tenant_id: Optional[int] = None,
+ code: Optional[str] = None,
+ user_name: Optional[str] = None,
):
super().__init__(name, description)
+ self.tenant_id = tenant_id
self.queue = queue
+ self.code = code
+ self.user_name = user_name
def create_if_not_exists(
self, queue_name: str, user=configuration.USER_NAME
) -> None:
"""Create Tenant if not exists."""
- JavaGate().create_tenant(self.name, queue_name, self.description)
+ tenant = JavaGate().create_tenant(self.name, self.description, queue_name)
+ self.tenant_id = tenant.getId()
+ self.code = tenant.getTenantCode()
# gateway_result_checker(result, None)
+
+ @classmethod
+ def get_tenant(cls, code: str) -> "Tenant":
+ """Get Tenant list."""
+ tenant = JavaGate().query_tenant(code)
+ if tenant is None:
+ return cls()
+ return cls(
+ description=tenant.getDescription(),
+ tenant_id=tenant.getId(),
+ code=tenant.getTenantCode(),
+ queue=tenant.getQueueId(),
+ )
+
+ def update(
+ self, user=configuration.USER_NAME, code=None, queue_id=None, description=None
+ ) -> None:
+ """Update Tenant."""
+ JavaGate().update_tenant(user, self.tenant_id, code, queue_id, description)
+ # TODO: check queue_id and queue_name
+ self.queue = str(queue_id)
+ self.code = code
+ self.description = description
+
+ def delete(self) -> None:
+ """Delete Tenant."""
+ JavaGate().delete_tenant(self.user_name, self.tenant_id)
+ self.delete_all()
diff --git a/src/pydolphinscheduler/models/user.py b/src/pydolphinscheduler/models/user.py
index e11bb9c..57c6af6 100644
--- a/src/pydolphinscheduler/models/user.py
+++ b/src/pydolphinscheduler/models/user.py
@@ -48,6 +48,7 @@ class User(BaseSide):
status: Optional[int] = configuration.USER_STATE,
):
super().__init__(name)
+ self.user_id: Optional[int] = None
self.password = password
self.email = email
self.phone = phone
@@ -64,7 +65,7 @@ class User(BaseSide):
"""Create User if not exists."""
# Should make sure queue already exists.
self.create_tenant_if_not_exists()
- JavaGate().create_user(
+ user = JavaGate().create_user(
self.name,
self.password,
self.email,
@@ -73,5 +74,57 @@ class User(BaseSide):
self.queue,
self.status,
)
+ self.user_id = user.getId()
# TODO recover result checker
# gateway_result_checker(result, None)
+
+ @classmethod
+ def get_user(cls, user_id) -> "User":
+ """Get User."""
+ user = JavaGate().query_user(user_id)
+ if user is None:
+ return cls("")
+ user_id = user.getId()
+ user = cls(
+ name=user.getUserName(),
+ password=user.getUserPassword(),
+ email=user.getEmail(),
+ phone=user.getPhone(),
+ tenant=user.getTenantCode(),
+ queue=user.getQueueName(),
+ status=user.getState(),
+ )
+ user.user_id = user_id
+ return user
+
+ def update(
+ self,
+ password=None,
+ email=None,
+ phone=None,
+ tenant=None,
+ queue=None,
+ status=None,
+ ) -> None:
+ """Update User."""
+ user = JavaGate().update_user(
+ self.name,
+ password,
+ email,
+ phone,
+ tenant,
+ queue,
+ status,
+ )
+ self.user_id = user.getId()
+ self.name = user.getUserName()
+ self.password = user.getUserPassword()
+ self.email = user.getEmail()
+ self.phone = user.getPhone()
+ self.queue = user.getQueueName()
+ self.status = user.getState()
+
+ def delete(self) -> None:
+ """Delete User."""
+ JavaGate().delete_user(self.name, self.user_id)
+ self.delete_all()
diff --git a/src/pydolphinscheduler/side/__init__.py b/src/pydolphinscheduler/resources_plugin/__init__.py
similarity index 63%
rename from src/pydolphinscheduler/side/__init__.py
rename to src/pydolphinscheduler/resources_plugin/__init__.py
index c4479dd..1e24e1e 100644
--- a/src/pydolphinscheduler/side/__init__.py
+++ b/src/pydolphinscheduler/resources_plugin/__init__.py
@@ -15,18 +15,11 @@
# specific language governing permissions and limitations
# under the License.
-"""Init Side package, Side package keep object related to DolphinScheduler but not in the Core part."""
+"""Init resources_plugin package."""
+from pydolphinscheduler.resources_plugin.github import GitHub
+from pydolphinscheduler.resources_plugin.gitlab import GitLab
+from pydolphinscheduler.resources_plugin.local import Local
+from pydolphinscheduler.resources_plugin.oss import OSS
+from pydolphinscheduler.resources_plugin.s3 import S3
-from pydolphinscheduler.side.project import Project
-from pydolphinscheduler.side.queue import Queue
-from pydolphinscheduler.side.tenant import Tenant
-from pydolphinscheduler.side.user import User
-from pydolphinscheduler.side.worker_group import WorkerGroup
-
-__all__ = [
- "Project",
- "Tenant",
- "User",
- "Queue",
- "WorkerGroup",
-]
+__all__ = ["Local", "GitHub", "GitLab", "OSS", "S3"]
diff --git a/src/pydolphinscheduler/side/worker_group.py b/src/pydolphinscheduler/resources_plugin/base/__init__.py
similarity index 68%
rename from src/pydolphinscheduler/side/worker_group.py
rename to src/pydolphinscheduler/resources_plugin/base/__init__.py
index ed50ec6..4253cda 100644
--- a/src/pydolphinscheduler/side/worker_group.py
+++ b/src/pydolphinscheduler/resources_plugin/base/__init__.py
@@ -15,16 +15,4 @@
# specific language governing permissions and limitations
# under the License.
-"""DolphinScheduler Worker Group object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core.base_side import BaseSide
-
-
-class WorkerGroup(BaseSide):
- """DolphinScheduler Worker Group object."""
-
- def __init__(self, name: str, address: str, description: Optional[str] = None):
- super().__init__(name, description)
- self.address = address
+"""Init base package."""
diff --git a/src/pydolphinscheduler/resources_plugin/base/bucket.py b/src/pydolphinscheduler/resources_plugin/base/bucket.py
new file mode 100644
index 0000000..bae4366
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/base/bucket.py
@@ -0,0 +1,86 @@
+# 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 BucketFileInfo and Bucket object."""
+from abc import ABCMeta, abstractmethod
+from typing import Optional
+
+
+class BucketFileInfo:
+ """A class that defines the details of BUCKET files.
+
+ :param bucket: A string representing the bucket to which the bucket file belongs.
+ :param file_path: A string representing the bucket file path.
+ """
+
+ def __init__(
+ self,
+ bucket: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ self.bucket = bucket
+ self.file_path = file_path
+
+
+class OSSFileInfo(BucketFileInfo):
+ """A class that defines the details of OSS files.
+
+ :param endpoint: A string representing the OSS file endpoint.
+ :param bucket: A string representing the bucket to which the OSS file belongs.
+ :param file_path: A string representing the OSS file path.
+ """
+
+ def __init__(
+ self,
+ endpoint: Optional[str] = None,
+ bucket: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ super().__init__(bucket=bucket, file_path=file_path, *args, **kwargs)
+ self.endpoint = endpoint
+
+
+class S3FileInfo(BucketFileInfo):
+ """A class that defines the details of S3 files.
+
+ :param bucket: A string representing the bucket to which the S3 file belongs.
+ :param file_path: A string representing the S3 file path.
+ """
+
+ def __init__(
+ self,
+ bucket: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ super().__init__(bucket=bucket, file_path=file_path, *args, **kwargs)
+
+
+class Bucket(object, metaclass=ABCMeta):
+ """An abstract class of online code repository based on git implementation."""
+
+ _bucket_file_info: Optional = None
+
+ @abstractmethod
+ def get_bucket_file_info(self, path: str):
+ """Get the detailed information of BUCKET file according to the file URL."""
+ raise NotImplementedError
diff --git a/src/pydolphinscheduler/resources_plugin/base/git.py b/src/pydolphinscheduler/resources_plugin/base/git.py
new file mode 100644
index 0000000..4fc2a17
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/base/git.py
@@ -0,0 +1,115 @@
+# 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 GitFileInfo and Git object."""
+
+from abc import ABCMeta, abstractmethod
+from typing import Optional
+
+
+class GitFileInfo:
+ """A class that defines the details of GIT files.
+
+ :param user: A string representing the user the git file belongs to.
+ :param repo_name: A string representing the repository to which the git file belongs.
+ :param branch: A string representing the branch to which the git file belongs.
+ :param file_path: A string representing the git file path.
+ """
+
+ def __init__(
+ self,
+ user: Optional[str] = None,
+ repo_name: Optional[str] = None,
+ branch: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ self.user = user
+ self.repo_name = repo_name
+ self.branch = branch
+ self.file_path = file_path
+
+
+class GitHubFileInfo(GitFileInfo):
+ """A class that defines the details of GitHub files.
+
+ :param user: A string representing the user the GitHub file belongs to.
+ :param repo_name: A string representing the repository to which the GitHub file belongs.
+ :param branch: A string representing the branch to which the GitHub file belongs.
+ :param file_path: A string representing the GitHub file path.
+ """
+
+ def __init__(
+ self,
+ user: Optional[str] = None,
+ repo_name: Optional[str] = None,
+ branch: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ super().__init__(
+ user=user,
+ repo_name=repo_name,
+ branch=branch,
+ file_path=file_path,
+ *args,
+ **kwargs
+ )
+
+
+class GitLabFileInfo(GitFileInfo):
+ """A class that defines the details of GitLab files.
+
+ :param host: A string representing the domain name the GitLab file belongs to.
+ :param user: A string representing the user the GitLab file belongs to.
+ :param repo_name: A string representing the repository to which the GitLab file belongs.
+ :param branch: A string representing the branch to which the GitHub file belongs.
+ :param file_path: A string representing the GitHub file path.
+ """
+
+ def __init__(
+ self,
+ host: Optional[str] = None,
+ user: Optional[str] = None,
+ repo_name: Optional[str] = None,
+ branch: Optional[str] = None,
+ file_path: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ super().__init__(
+ user=user,
+ repo_name=repo_name,
+ branch=branch,
+ file_path=file_path,
+ *args,
+ **kwargs
+ )
+ self.host = host
+
+
+class Git(object, metaclass=ABCMeta):
+ """An abstract class of online code repository based on git implementation."""
+
+ _git_file_info: Optional = None
+
+ @abstractmethod
+ def get_git_file_info(self, path: str):
+ """Get the detailed information of GIT file according to the file URL."""
+ raise NotImplementedError
diff --git a/src/pydolphinscheduler/resources_plugin/github.py b/src/pydolphinscheduler/resources_plugin/github.py
new file mode 100644
index 0000000..4564864
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/github.py
@@ -0,0 +1,106 @@
+# 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 github resource plugin."""
+import base64
+from typing import Optional
+from urllib.parse import urljoin
+
+import requests
+
+from pydolphinscheduler.constants import Symbol
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.resources_plugin.base.git import Git, GitHubFileInfo
+
+
+class GitHub(ResourcePlugin, Git):
+ """GitHub resource plugin, a plugin for task and workflow to dolphinscheduler to read resource.
+
+ :param prefix: A string representing the prefix of GitHub.
+ :param access_token: A string used for identity authentication of GitHub private repository.
+ """
+
+ def __init__(
+ self, prefix: str, access_token: Optional[str] = None, *args, **kwargs
+ ):
+ super().__init__(prefix, *args, **kwargs)
+ self.access_token = access_token
+
+ _git_file_info: Optional[GitHubFileInfo] = None
+
+ def build_req_api(
+ self,
+ user: str,
+ repo_name: str,
+ file_path: str,
+ api: str,
+ ):
+ """Build request file content API."""
+ api = api.replace("{user}", user)
+ api = api.replace("{repo_name}", repo_name)
+ api = api.replace("{file_path}", file_path)
+ return api
+
+ def get_git_file_info(self, path: str):
+ """Get file information from the file url, like repository name, user, branch, and file path."""
+ elements = path.split(Symbol.SLASH)
+ index = self.get_index(path, Symbol.SLASH, 7)
+ index = index + 1
+ file_info = GitHubFileInfo(
+ user=elements[3],
+ repo_name=elements[4],
+ branch=elements[6],
+ file_path=path[index:],
+ )
+ self._git_file_info = file_info
+
+ def get_req_url(self):
+ """Build request URL according to file information."""
+ return self.build_req_api(
+ user=self._git_file_info.user,
+ repo_name=self._git_file_info.repo_name,
+ file_path=self._git_file_info.file_path,
+ api="https://api.github.com/repos/{user}/{repo_name}/contents/{file_path}",
+ )
+
+ 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.
+ """
+ path = urljoin(self.prefix, suf)
+ return self.req(path)
+
+ def req(self, path: str):
+ """Send HTTP request, parse response data, and get file content."""
+ headers = {
+ "Content-Type": "application/json; charset=utf-8",
+ }
+ if self.access_token is not None:
+ headers.setdefault("Authorization", "Bearer %s" % self.access_token)
+ self.get_git_file_info(path)
+ response = requests.get(
+ headers=headers,
+ url=self.get_req_url(),
+ params={"ref": self._git_file_info.branch},
+ )
+ if response.status_code == requests.codes.ok:
+ json_response = response.json()
+ content = base64.b64decode(json_response["content"])
+ return content.decode("utf-8")
+ else:
+ raise Exception(response.json())
diff --git a/src/pydolphinscheduler/resources_plugin/gitlab.py b/src/pydolphinscheduler/resources_plugin/gitlab.py
new file mode 100644
index 0000000..f035eca
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/gitlab.py
@@ -0,0 +1,112 @@
+# 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 gitlab resource plugin."""
+from typing import Optional
+from urllib.parse import urljoin, urlparse
+
+import gitlab
+import requests
+
+from pydolphinscheduler.constants import Symbol
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.resources_plugin.base.git import Git, GitLabFileInfo
+
+
+class GitLab(ResourcePlugin, Git):
+ """GitLab object, declare GitLab resource plugin for task and workflow to dolphinscheduler.
+
+ :param prefix: A string representing the prefix of GitLab.
+ :param private_token: A string used for identity authentication of GitLab private or Internal repository.
+ :param oauth_token: A string used for identity authentication of GitLab private or Internal repository.
+ :param username: A string representing the user of the repository.
+ :param password: A string representing the user password.
+ """
+
+ def __init__(
+ self,
+ prefix: str,
+ private_token: Optional[str] = None,
+ oauth_token: Optional[str] = None,
+ username: Optional[str] = None,
+ password: Optional[str] = None,
+ *args,
+ **kwargs,
+ ):
+ super().__init__(prefix, *args, **kwargs)
+ self.private_token = private_token
+ self.oauth_token = oauth_token
+ self.username = username
+ self.password = password
+
+ def get_git_file_info(self, path: str):
+ """Get file information from the file url, like repository name, user, branch, and file path."""
+ self.get_index(path, Symbol.SLASH, 8)
+ result = urlparse(path)
+ elements = result.path.split(Symbol.SLASH)
+ self._git_file_info = GitLabFileInfo(
+ host=f"{result.scheme}://{result.hostname}",
+ repo_name=elements[2],
+ branch=elements[5],
+ file_path=Symbol.SLASH.join(
+ str(elements[i]) for i in range(6, len(elements))
+ ),
+ user=elements[1],
+ )
+
+ def authentication(self):
+ """Gitlab authentication."""
+ host = self._git_file_info.host
+ if self.private_token is not None:
+ return gitlab.Gitlab(host, private_token=self.private_token)
+ if self.oauth_token is not None:
+ return gitlab.Gitlab(host, oauth_token=self.oauth_token)
+ if self.username is not None and self.password is not None:
+ oauth_token = self.OAuth_token()
+ return gitlab.Gitlab(host, oauth_token=oauth_token)
+ return gitlab.Gitlab(host)
+
+ def OAuth_token(self):
+ """Obtain OAuth Token."""
+ data = {
+ "grant_type": "password",
+ "username": self.username,
+ "password": self.password,
+ }
+ host = self._git_file_info.host
+ resp = requests.post("%s/oauth/token" % host, data=data)
+ oauth_token = resp.json()["access_token"]
+ return oauth_token
+
+ 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.
+ """
+ path = urljoin(self.prefix, suf)
+ self.get_git_file_info(path)
+ gl = self.authentication()
+ project = gl.projects.get(
+ "%s/%s" % (self._git_file_info.user, self._git_file_info.repo_name)
+ )
+ return (
+ project.files.get(
+ file_path=self._git_file_info.file_path, ref=self._git_file_info.branch
+ )
+ .decode()
+ .decode()
+ )
diff --git a/src/pydolphinscheduler/resources_plugin/local.py b/src/pydolphinscheduler/resources_plugin/local.py
new file mode 100644
index 0000000..c1fc56d
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/local.py
@@ -0,0 +1,56 @@
+# 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 local resource plugin."""
+
+import os
+from pathlib import Path
+
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.exceptions import PyResPluginException
+
+
+class Local(ResourcePlugin):
+ """Local object, declare local resource plugin for task and workflow to dolphinscheduler.
+
+ :param prefix: A string representing the prefix of Local.
+ """
+
+ # [start init_method]
+ def __init__(self, prefix: str, *args, **kwargs):
+ super().__init__(prefix, *args, **kwargs)
+
+ # [end init_method]
+
+ # [start read_file_method]
+ 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.
+ """
+ path = Path(self.prefix).joinpath(suf)
+ if not path.exists():
+ raise PyResPluginException("{} is not found".format(str(path)))
+ if not os.access(str(path), os.R_OK):
+ raise PyResPluginException(
+ "You don't have permission to access {}".format(self.prefix + suf)
+ )
+ with open(path, "r") as f:
+ content = f.read()
+ return content
+
+ # [end read_file_method]
diff --git a/src/pydolphinscheduler/resources_plugin/oss.py b/src/pydolphinscheduler/resources_plugin/oss.py
new file mode 100644
index 0000000..1a9acbb
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/oss.py
@@ -0,0 +1,76 @@
+# 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 oss resource plugin."""
+from typing import Optional
+from urllib.parse import urljoin, urlparse
+
+import oss2
+
+from pydolphinscheduler.constants import Symbol
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.resources_plugin.base.bucket import Bucket, OSSFileInfo
+
+
+class OSS(ResourcePlugin, Bucket):
+ """OSS object, declare OSS resource plugin for task and workflow to dolphinscheduler.
+
+ :param prefix: A string representing the prefix of OSS.
+ :param access_key_id: A string representing the ID of AccessKey for AliCloud OSS.
+ :param access_key_secret: A string representing the secret of AccessKey for AliCloud OSS.
+ """
+
+ def __init__(
+ self,
+ prefix: str,
+ access_key_id: Optional[str] = None,
+ access_key_secret: Optional[str] = None,
+ *args,
+ **kwargs,
+ ):
+ super().__init__(prefix, *args, **kwargs)
+ self.access_key_id = access_key_id
+ self.access_key_secret = access_key_secret
+
+ _bucket_file_info: Optional[OSSFileInfo] = None
+
+ def get_bucket_file_info(self, path: str):
+ """Get file information from the file url, like repository name, user, branch, and file path."""
+ self.get_index(path, Symbol.SLASH, 3)
+ result = urlparse(path)
+ hostname = result.hostname
+ elements = hostname.split(Symbol.POINT)
+ self._bucket_file_info = OSSFileInfo(
+ endpoint=f"{result.scheme}://"
+ f"{Symbol.POINT.join(str(elements[i]) for i in range(1, len(elements)))}",
+ bucket=hostname.split(Symbol.POINT)[0],
+ file_path=result.path[1:],
+ )
+
+ 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.
+ """
+ path = urljoin(self.prefix, suf)
+ self.get_bucket_file_info(path)
+ auth = oss2.Auth(self.access_key_id, self.access_key_secret)
+ bucket = oss2.Bucket(
+ auth, self._bucket_file_info.endpoint, self._bucket_file_info.bucket
+ )
+ result = bucket.get_object(self._bucket_file_info.file_path).read().decode()
+ return result.read().decode()
diff --git a/src/pydolphinscheduler/resources_plugin/s3.py b/src/pydolphinscheduler/resources_plugin/s3.py
new file mode 100644
index 0000000..da1fe83
--- /dev/null
+++ b/src/pydolphinscheduler/resources_plugin/s3.py
@@ -0,0 +1,74 @@
+# 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 S3 resource plugin."""
+
+from typing import Optional
+from urllib.parse import urljoin
+
+import boto3
+
+from pydolphinscheduler.constants import Symbol
+from pydolphinscheduler.core.resource_plugin import ResourcePlugin
+from pydolphinscheduler.resources_plugin.base.bucket import Bucket, S3FileInfo
+
+
+class S3(ResourcePlugin, Bucket):
+ """S3 object, declare S3 resource plugin for task and workflow to dolphinscheduler.
+
+ :param prefix: A string representing the prefix of S3.
+ :param access_key_id: A string representing the ID of AccessKey for Amazon S3.
+ :param access_key_secret: A string representing the secret of AccessKey for Amazon S3.
+ """
+
+ def __init__(
+ self,
+ prefix: str,
+ access_key_id: Optional[str] = None,
+ access_key_secret: Optional[str] = None,
+ *args,
+ **kwargs
+ ):
+ super().__init__(prefix, *args, **kwargs)
+ self.access_key_id = access_key_id
+ self.access_key_secret = access_key_secret
+
+ _bucket_file_info: Optional[S3FileInfo] = None
+
+ def get_bucket_file_info(self, path: str):
+ """Get file information from the file url, like repository name, user, branch, and file path."""
+ elements = path.split(Symbol.SLASH)
+ self.get_index(path, Symbol.SLASH, 3)
+ self._bucket_file_info = S3FileInfo(
+ bucket=elements[2].split(Symbol.POINT)[0],
+ file_path=Symbol.SLASH.join(
+ str(elements[i]) for i in range(3, len(elements))
+ ),
+ )
+
+ 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.
+ """
+ path = urljoin(self.prefix, suf)
+ self.get_bucket_file_info(path)
+ bucket = self._bucket_file_info.bucket
+ key = self._bucket_file_info.file_path
+ s3_resource = boto3.resource("s3")
+ s3_object = s3_resource.Object(bucket, key)
+ return s3_object.get()["Body"].read().decode("utf-8")
diff --git a/src/pydolphinscheduler/side/project.py b/src/pydolphinscheduler/side/project.py
deleted file mode 100644
index 750e3b8..0000000
--- a/src/pydolphinscheduler/side/project.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# 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 Project object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core import configuration
-from pydolphinscheduler.core.base_side import BaseSide
-from pydolphinscheduler.java_gateway import launch_gateway
-
-
-class Project(BaseSide):
- """DolphinScheduler Project object."""
-
- def __init__(
- self,
- name: str = configuration.WORKFLOW_PROJECT,
- description: Optional[str] = None,
- ):
- super().__init__(name, description)
-
- def create_if_not_exists(self, user=configuration.USER_NAME) -> None:
- """Create Project if not exists."""
- gateway = launch_gateway()
- gateway.entry_point.createOrGrantProject(user, self.name, self.description)
- # TODO recover result checker
- # gateway_result_checker(result, None)
diff --git a/src/pydolphinscheduler/side/queue.py b/src/pydolphinscheduler/side/queue.py
deleted file mode 100644
index e7c68e1..0000000
--- a/src/pydolphinscheduler/side/queue.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# 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 User object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core import configuration
-from pydolphinscheduler.core.base_side import BaseSide
-from pydolphinscheduler.java_gateway import gateway_result_checker, launch_gateway
-
-
-class Queue(BaseSide):
- """DolphinScheduler Queue object."""
-
- def __init__(
- self,
- name: str = configuration.WORKFLOW_QUEUE,
- description: Optional[str] = "",
- ):
- super().__init__(name, description)
-
- def create_if_not_exists(self, user=configuration.USER_NAME) -> None:
- """Create Queue if not exists."""
- gateway = launch_gateway()
- # Here we set Queue.name and Queue.queueName same as self.name
- result = gateway.entry_point.createProject(user, self.name, self.name)
- gateway_result_checker(result, None)
diff --git a/src/pydolphinscheduler/side/tenant.py b/src/pydolphinscheduler/side/tenant.py
deleted file mode 100644
index 6aaabfe..0000000
--- a/src/pydolphinscheduler/side/tenant.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# 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 Tenant object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core import configuration
-from pydolphinscheduler.core.base_side import BaseSide
-from pydolphinscheduler.java_gateway import launch_gateway
-
-
-class Tenant(BaseSide):
- """DolphinScheduler Tenant object."""
-
- def __init__(
- self,
- name: str = configuration.WORKFLOW_TENANT,
- queue: str = configuration.WORKFLOW_QUEUE,
- description: Optional[str] = None,
- ):
- super().__init__(name, description)
- self.queue = queue
-
- def create_if_not_exists(
- self, queue_name: str, user=configuration.USER_NAME
- ) -> None:
- """Create Tenant if not exists."""
- gateway = launch_gateway()
- gateway.entry_point.createTenant(self.name, self.description, queue_name)
- # gateway_result_checker(result, None)
diff --git a/src/pydolphinscheduler/side/user.py b/src/pydolphinscheduler/side/user.py
deleted file mode 100644
index 510e3a8..0000000
--- a/src/pydolphinscheduler/side/user.py
+++ /dev/null
@@ -1,79 +0,0 @@
-# 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 User object."""
-
-from typing import Optional
-
-from pydolphinscheduler.core import configuration
-from pydolphinscheduler.core.base_side import BaseSide
-from pydolphinscheduler.java_gateway import launch_gateway
-from pydolphinscheduler.side.tenant import Tenant
-
-
-class User(BaseSide):
- """DolphinScheduler User object."""
-
- _KEY_ATTR = {
- "name",
- "password",
- "email",
- "phone",
- "tenant",
- "queue",
- "status",
- }
-
- def __init__(
- self,
- name: str,
- password: Optional[str] = configuration.USER_PASSWORD,
- email: Optional[str] = configuration.USER_EMAIL,
- phone: Optional[str] = configuration.USER_PHONE,
- tenant: Optional[str] = configuration.WORKFLOW_TENANT,
- queue: Optional[str] = configuration.WORKFLOW_QUEUE,
- status: Optional[int] = configuration.USER_STATE,
- ):
- super().__init__(name)
- self.password = password
- self.email = email
- self.phone = phone
- self.tenant = tenant
- self.queue = queue
- self.status = status
-
- def create_tenant_if_not_exists(self) -> None:
- """Create tenant object."""
- tenant = Tenant(name=self.tenant, queue=self.queue)
- tenant.create_if_not_exists(self.queue)
-
- def create_if_not_exists(self, **kwargs):
- """Create User if not exists."""
- # Should make sure queue already exists.
- self.create_tenant_if_not_exists()
- gateway = launch_gateway()
- gateway.entry_point.createUser(
- self.name,
- self.password,
- self.email,
- self.phone,
- self.tenant,
- self.queue,
- self.status,
- )
- # TODO recover result checker
- # gateway_result_checker(result, None)
diff --git a/src/pydolphinscheduler/tasks/datax.py b/src/pydolphinscheduler/tasks/datax.py
index f881a67..945f782 100644
--- a/src/pydolphinscheduler/tasks/datax.py
+++ b/src/pydolphinscheduler/tasks/datax.py
@@ -34,6 +34,9 @@ class CustomDataX(Task):
_task_custom_attr = {"custom_config", "json", "xms", "xmx"}
+ ext: set = {".json"}
+ ext_attr: str = "_json"
+
def __init__(
self,
name: str,
@@ -43,9 +46,9 @@ class CustomDataX(Task):
*args,
**kwargs
):
+ self._json = json
super().__init__(name, TaskType.DATAX, *args, **kwargs)
self.custom_config = self.CUSTOM_CONFIG
- self.json = json
self.xms = xms
self.xmx = xmx
@@ -76,6 +79,9 @@ class DataX(Task):
"xmx",
}
+ ext: set = {".sql"}
+ ext_attr: str = "_sql"
+
def __init__(
self,
name: str,
@@ -92,8 +98,8 @@ class DataX(Task):
*args,
**kwargs
):
+ self._sql = sql
super().__init__(name, TaskType.DATAX, *args, **kwargs)
- self.sql = sql
self.custom_config = self.CUSTOM_CONFIG
self.datasource_name = datasource_name
self.datatarget_name = datatarget_name
diff --git a/src/pydolphinscheduler/tasks/python.py b/src/pydolphinscheduler/tasks/python.py
index 52903d4..593cc52 100644
--- a/src/pydolphinscheduler/tasks/python.py
+++ b/src/pydolphinscheduler/tasks/python.py
@@ -33,37 +33,38 @@ log = logging.getLogger(__file__)
class Python(Task):
"""Task Python object, declare behavior for Python task to dolphinscheduler.
- Python task support two types of parameters for :param:``code``, and here is an example:
+ Python task support two types of parameters for :param:``definition``, and here is an example:
- Using str type of :param:``code``
+ Using str type of :param:``definition``
.. code-block:: python
- python_task = Python(name="str_type", code="print('Hello Python task.')")
+ python_task = Python(name="str_type", definition="print('Hello Python task.')")
- Or using Python callable type of :param:``code``
+ Or using Python callable type of :param:``definition``
.. code-block:: python
def foo():
print("Hello Python task.")
- python_task = Python(name="str_type", code=foo)
+ python_task = Python(name="str_type", definition=foo)
:param name: The name for Python task. It define the task name.
:param definition: String format of Python script you want to execute or Python callable you
want to execute.
"""
- _task_custom_attr = {
- "raw_script",
- }
+ _task_custom_attr = {"raw_script", "definition"}
+
+ ext: set = {".py"}
+ ext_attr: Union[str, types.FunctionType] = "_definition"
def __init__(
self, name: str, definition: Union[str, types.FunctionType], *args, **kwargs
):
+ self._definition = definition
super().__init__(name, TaskType.PYTHON, *args, **kwargs)
- self.definition = definition
def _build_exe_str(self) -> str:
"""Build executable string from given definition.
@@ -71,32 +72,34 @@ class Python(Task):
Attribute ``self.definition`` almost is a function, we need to call this function after parsing it
to string. The easier way to call a function is using syntax ``func()`` and we use it to call it too.
"""
- if isinstance(self.definition, types.FunctionType):
- py_function = inspect.getsource(self.definition)
- func_str = f"{py_function}{self.definition.__name__}()"
+ definition = getattr(self, "definition")
+ if isinstance(definition, types.FunctionType):
+ py_function = inspect.getsource(definition)
+ func_str = f"{py_function}{definition.__name__}()"
else:
pattern = re.compile("^def (\\w+)\\(")
- find = pattern.findall(self.definition)
+ find = pattern.findall(definition)
if not find:
log.warning(
"Python definition is simple script instead of function, with value %s",
- self.definition,
+ definition,
)
- return self.definition
+ return definition
# Keep function str and function callable always have one blank line
func_str = (
- f"{self.definition}{find[0]}()"
- if self.definition.endswith("\n")
- else f"{self.definition}\n{find[0]}()"
+ f"{definition}{find[0]}()"
+ if definition.endswith("\n")
+ else f"{definition}\n{find[0]}()"
)
return func_str
@property
def raw_script(self) -> str:
"""Get python task define attribute `raw_script`."""
- if isinstance(self.definition, (str, types.FunctionType)):
+ if isinstance(getattr(self, "definition"), (str, types.FunctionType)):
return self._build_exe_str()
else:
raise PyDSParamException(
- "Parameter definition do not support % for now.", type(self.definition)
+ "Parameter definition do not support % for now.",
+ type(getattr(self, "definition")),
)
diff --git a/src/pydolphinscheduler/tasks/shell.py b/src/pydolphinscheduler/tasks/shell.py
index 9a73535..36ec4e8 100644
--- a/src/pydolphinscheduler/tasks/shell.py
+++ b/src/pydolphinscheduler/tasks/shell.py
@@ -50,6 +50,9 @@ class Shell(Task):
"raw_script",
}
+ ext: set = {".sh", ".zsh"}
+ ext_attr: str = "_raw_script"
+
def __init__(self, name: str, command: str, *args, **kwargs):
+ self._raw_script = command
super().__init__(name, TaskType.SHELL, *args, **kwargs)
- self.raw_script = command
diff --git a/src/pydolphinscheduler/tasks/spark.py b/src/pydolphinscheduler/tasks/spark.py
index 565daad..eb9c621 100644
--- a/src/pydolphinscheduler/tasks/spark.py
+++ b/src/pydolphinscheduler/tasks/spark.py
@@ -23,13 +23,6 @@ from pydolphinscheduler.constants import TaskType
from pydolphinscheduler.core.engine import Engine, ProgramType
-class SparkVersion(str):
- """Spark version, for now it just contain `SPARK1` and `SPARK2`."""
-
- SPARK1 = "SPARK1"
- SPARK2 = "SPARK2"
-
-
class DeployMode(str):
"""SPARK deploy mode, for now it just contain `LOCAL`, `CLIENT` and `CLUSTER`."""
@@ -43,7 +36,6 @@ class Spark(Engine):
_task_custom_attr = {
"deploy_mode",
- "spark_version",
"driver_cores",
"driver_memory",
"num_executors",
@@ -61,7 +53,6 @@ class Spark(Engine):
main_package: str,
program_type: Optional[ProgramType] = ProgramType.SCALA,
deploy_mode: Optional[DeployMode] = DeployMode.CLUSTER,
- spark_version: Optional[SparkVersion] = SparkVersion.SPARK2,
app_name: Optional[str] = None,
driver_cores: Optional[int] = 1,
driver_memory: Optional[str] = "512M",
@@ -83,7 +74,6 @@ class Spark(Engine):
**kwargs
)
self.deploy_mode = deploy_mode
- self.spark_version = spark_version
self.app_name = app_name
self.driver_cores = driver_cores
self.driver_memory = driver_memory
diff --git a/src/pydolphinscheduler/tasks/sql.py b/src/pydolphinscheduler/tasks/sql.py
index 716a024..4bebf83 100644
--- a/src/pydolphinscheduler/tasks/sql.py
+++ b/src/pydolphinscheduler/tasks/sql.py
@@ -59,6 +59,9 @@ class Sql(Task):
"display_rows",
}
+ ext: set = {".sql"}
+ ext_attr: str = "_sql"
+
def __init__(
self,
name: str,
@@ -71,8 +74,8 @@ class Sql(Task):
*args,
**kwargs
):
+ self._sql = sql
super().__init__(name, TaskType.SQL, *args, **kwargs)
- self.sql = sql
self.param_sql_type = sql_type
self.datasource_name = datasource_name
self.pre_statements = pre_statements or []
@@ -101,7 +104,7 @@ class Sql(Task):
"|(.* |)update |(.* |)truncate |(.* |)alter |(.* |)create ).*"
)
pattern_select = re.compile(pattern_select_str, re.IGNORECASE)
- if pattern_select.match(self.sql) is None:
+ if pattern_select.match(self._sql) is None:
return SqlType.NOT_SELECT
else:
return SqlType.SELECT
diff --git a/tests/core/test_task.py b/tests/core/test_task.py
index 87ebc99..c6ef777 100644
--- a/tests/core/test_task.py
+++ b/tests/core/test_task.py
@@ -18,14 +18,15 @@
"""Test Task class function."""
import logging
import re
-from unittest.mock import patch
from typing import Set
-from unittest.mock import patch
+from unittest.mock import PropertyMock, patch
import pytest
from pydolphinscheduler.core.process_definition import ProcessDefinition
from pydolphinscheduler.core.task import Task, TaskRelation
+from pydolphinscheduler.exceptions import PyResPluginException
+from pydolphinscheduler.resources_plugin import Local
from tests.testing.task import Task as TestTask
from tests.testing.task import TaskWithCode
@@ -320,6 +321,118 @@ def test_add_duplicate(caplog):
)
+@pytest.mark.parametrize(
+ "val, expected",
+ [
+ ("a.sh", "echo Test task attribute ext_attr"),
+ ("a.zsh", "echo Test task attribute ext_attr"),
+ ("echo Test task attribute ext_attr", "echo Test task attribute ext_attr"),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.ext",
+ new_callable=PropertyMock,
+ return_value={".sh", ".zsh"},
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.ext_attr",
+ new_callable=PropertyMock,
+ return_value="_raw_script",
+)
+@patch(
+ "pydolphinscheduler.core.task.Task._raw_script",
+ create=True,
+ new_callable=PropertyMock,
+)
+@patch("pydolphinscheduler.core.task.Task.get_plugin")
+def test_task_ext_attr(
+ m_plugin, m_raw_script, m_ext_attr, m_ext, m_code_version, val, expected
+):
+ """Test task attribute ext_attr."""
+ m_plugin.return_value.read_file.return_value = expected
+ m_raw_script.return_value = val
+ task = Task("test_task_ext_attr", "test_task_ext_attr")
+ assert expected == getattr(task, "raw_script")
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "name": "test_task_abtain_res_plugin",
+ "task_type": "TaskType",
+ "resource_plugin": Local("prefix"),
+ "process_definition": ProcessDefinition(
+ name="process_definition",
+ resource_plugin=Local("prefix"),
+ ),
+ },
+ "Local",
+ ),
+ (
+ {
+ "name": "test_task_abtain_res_plugin",
+ "task_type": "TaskType",
+ "resource_plugin": Local("prefix"),
+ },
+ "Local",
+ ),
+ (
+ {
+ "name": "test_task_abtain_res_plugin",
+ "task_type": "TaskType",
+ "process_definition": ProcessDefinition(
+ name="process_definition",
+ resource_plugin=Local("prefix"),
+ ),
+ },
+ "Local",
+ ),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+@patch("pydolphinscheduler.core.task.Task.get_content")
+def test_task_obtain_res_plugin(m_get_content, m_code_version, attr, expected):
+ """Test task obtaining resource plug-in."""
+ task = Task(**attr)
+ assert expected == task.get_plugin().__class__.__name__
+
+
+@pytest.mark.parametrize(
+ "attr",
+ [
+ {
+ "name": "test_task_abtain_res_plugin",
+ "task_type": "TaskType",
+ "process_definition": ProcessDefinition(
+ name="process_definition",
+ ),
+ },
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+@patch("pydolphinscheduler.core.task.Task.get_content")
+def test_task_obtain_res_plugin_exception(m_get_content, m_code_version, attr):
+ """Test task obtaining resource plug-in exception."""
+ with pytest.raises(
+ PyResPluginException,
+ match="The execution command of this task is a file, but the resource plugin is empty",
+ ):
+ task = Task(**attr)
+ task.get_plugin()
+
+
@pytest.mark.parametrize(
"resources, expect",
[
diff --git a/tests/example/test_example.py b/tests/example/test_example.py
index 70f3677..319ad96 100644
--- a/tests/example/test_example.py
+++ b/tests/example/test_example.py
@@ -97,7 +97,11 @@ def test_example_basic():
), f"We expect all examples is python script, but get {ex.name}."
# All except tutorial and __init__ is end with keyword "_example"
- if ex.stem not in ("tutorial", "tutorial_decorator") and ex.stem != "__init__":
+ if (
+ ex.stem
+ not in ("tutorial", "tutorial_decorator", "tutorial_resource_plugin")
+ and ex.stem != "__init__"
+ ):
assert ex.stem.endswith(
"_example"
), f"We expect all examples script end with keyword '_example', but get {ex.stem}."
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 236956b..c15b897 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -42,7 +42,7 @@ def docker_setup_teardown():
image="apache/dolphinscheduler-standalone-server:ci",
container_name="ci-dolphinscheduler-standalone-server",
)
- ports = {"25333/tcp": 25333}
+ ports = {"25333/tcp": 25333, "12345/tcp": 12345}
container = docker_wrapper.run_until_log(
log="Started StandaloneServer in", tty=True, ports=ports
)
diff --git a/tests/integration/test_project.py b/tests/integration/test_project.py
new file mode 100644
index 0000000..167ce2d
--- /dev/null
+++ b/tests/integration/test_project.py
@@ -0,0 +1,78 @@
+# 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.
+
+"""Test pydolphinscheduler project."""
+import pytest
+
+from pydolphinscheduler.models import Project, User
+
+
+def get_user(
+ name="test-name",
+ password="test-password",
+ email="test-email@abc.com",
+ phone="17366637777",
+ tenant="test-tenant",
+ queue="test-queue",
+ status=1,
+):
+ """Get a test user."""
+ user = User(name, password, email, phone, tenant, queue, status)
+ user.create_if_not_exists()
+ return user
+
+
+def get_project(name="test-name-1", description="test-description", code=1):
+ """Get a test project."""
+ project = Project(name, description, code=code)
+ user = get_user()
+ project.create_if_not_exists(user=user.name)
+ return project
+
+
+def test_create_and_get_project():
+ """Test create and get project from java gateway."""
+ project = get_project()
+ project_ = Project.get_project_by_name(user="test-name", name=project.name)
+ assert project_.name == project.name
+ assert project_.description == project.description
+
+
+def test_update_project():
+ """Test update project from java gateway."""
+ project = get_project()
+ project = project.get_project_by_name(user="test-name", name=project.name)
+ project.update(
+ user="test-name",
+ project_code=project.code,
+ project_name="test-name-updated",
+ description="test-description-updated",
+ )
+ project_ = Project.get_project_by_name(user="test-name", name="test-name-updated")
+ assert project_.description == "test-description-updated"
+
+
+def test_delete_project():
+ """Test delete project from java gateway."""
+ project = get_project()
+ project.get_project_by_name(user="test-name", name=project.name)
+ project.delete(user="test-name")
+
+ with pytest.raises(AttributeError) as excinfo:
+ _ = project.name
+
+ assert excinfo.type == AttributeError
diff --git a/tests/integration/test_tenant.py b/tests/integration/test_tenant.py
new file mode 100644
index 0000000..c1ec33c
--- /dev/null
+++ b/tests/integration/test_tenant.py
@@ -0,0 +1,86 @@
+# 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.
+
+"""Test pydolphinscheduler tenant."""
+import pytest
+
+from pydolphinscheduler.models import Tenant, User
+
+
+def get_user(
+ name="test-name",
+ password="test-password",
+ email="test-email@abc.com",
+ phone="17366637777",
+ tenant="test-tenant",
+ queue="test-queue",
+ status=1,
+):
+ """Get a test user."""
+ user = User(name, password, email, phone, tenant, queue, status)
+ user.create_if_not_exists()
+ return user
+
+
+def get_tenant(
+ name="test-name-1",
+ queue="test-queue-1",
+ description="test-description",
+ tenant_code="test-tenant-code",
+ user_name=None,
+):
+ """Get a test tenant."""
+ tenant = Tenant(name, queue, description, code=tenant_code, user_name=user_name)
+ tenant.create_if_not_exists(name)
+ return tenant
+
+
+def test_create_tenant():
+ """Test create tenant from java gateway."""
+ tenant = get_tenant()
+ assert tenant.tenant_id is not None
+
+
+def test_get_tenant():
+ """Test get tenant from java gateway."""
+ tenant = get_tenant()
+ tenant_ = Tenant.get_tenant(tenant.code)
+ assert tenant_.tenant_id == tenant.tenant_id
+
+
+def test_update_tenant():
+ """Test update tenant from java gateway."""
+ tenant = get_tenant(user_name="admin")
+ tenant.update(
+ user="admin",
+ code="test-code-updated",
+ queue_id=1,
+ description="test-description-updated",
+ )
+ tenant_ = Tenant.get_tenant(code=tenant.code)
+ assert tenant_.code == "test-code-updated"
+ assert tenant_.queue == 1
+
+
+def test_delete_tenant():
+ """Test delete tenant from java gateway."""
+ tenant = get_tenant(user_name="admin")
+ tenant.delete()
+ with pytest.raises(AttributeError) as excinfo:
+ _ = tenant.tenant_id
+
+ assert excinfo.type == AttributeError
diff --git a/tests/integration/test_user.py b/tests/integration/test_user.py
new file mode 100644
index 0000000..74248fa
--- /dev/null
+++ b/tests/integration/test_user.py
@@ -0,0 +1,107 @@
+# 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.
+
+"""Test pydolphinscheduler user."""
+
+import hashlib
+
+import pytest
+
+from pydolphinscheduler.models import User
+
+
+def md5(str):
+ """MD5 a string."""
+ hl = hashlib.md5()
+ hl.update(str.encode(encoding="utf-8"))
+ return hl.hexdigest()
+
+
+def get_user(
+ name="test-name",
+ password="test-password",
+ email="test-email@abc.com",
+ phone="17366637777",
+ tenant="test-tenant",
+ queue="test-queue",
+ status=1,
+):
+ """Get a test user."""
+ user = User(
+ name=name,
+ password=password,
+ email=email,
+ phone=phone,
+ tenant=tenant,
+ queue=queue,
+ status=status,
+ )
+ user.create_if_not_exists()
+ return user
+
+
+def test_create_user():
+ """Test weather client could connect java gate way or not."""
+ user = User(
+ name="test-name",
+ password="test-password",
+ email="test-email@abc.com",
+ phone="17366637777",
+ tenant="test-tenant",
+ queue="test-queue",
+ status=1,
+ )
+ user.create_if_not_exists()
+ assert user.user_id is not None
+
+
+def test_get_user():
+ """Test get user from java gateway."""
+ user = get_user()
+ user_ = User.get_user(user.user_id)
+ assert user_.password == md5(user.password)
+ assert user_.email == user.email
+ assert user_.phone == user.phone
+ assert user_.status == user.status
+
+
+def test_update_user():
+ """Test update user from java gateway."""
+ user = get_user()
+ user.update(
+ password="test-password-",
+ email="test-email-updated@abc.com",
+ phone="17366637766",
+ tenant="test-tenant-updated",
+ queue="test-queue-updated",
+ status=2,
+ )
+ user_ = User.get_user(user.user_id)
+ assert user_.password == md5("test-password-")
+ assert user_.email == "test-email-updated@abc.com"
+ assert user_.phone == "17366637766"
+ assert user_.status == 2
+
+
+def test_delete_user():
+ """Test delete user from java gateway."""
+ user = get_user()
+ user.delete()
+ with pytest.raises(AttributeError) as excinfo:
+ _ = user.user_id
+
+ assert excinfo.type == AttributeError
diff --git a/examples/yaml_define/Spark.yaml b/tests/resources_plugin/__init__.py
similarity index 73%
copy from examples/yaml_define/Spark.yaml
copy to tests/resources_plugin/__init__.py
index 6132b8d..0b6bdf3 100644
--- a/examples/yaml_define/Spark.yaml
+++ b/tests/resources_plugin/__init__.py
@@ -15,16 +15,4 @@
# specific language governing permissions and limitations
# under the License.
-# Define the workflow
-workflow:
- name: "Spark"
-
-# Define the tasks under the workflow
-tasks:
- - name: task
- task_type: Spark
- main_class: org.apache.spark.examples.SparkPi
- main_package: test_java.jar
- program_type: SCALA
- deploy_mode: local
- spark_version: SPARK1
+"""Init resources_plugin package tests."""
diff --git a/tests/resources_plugin/test_github.py b/tests/resources_plugin/test_github.py
new file mode 100644
index 0000000..1f1a631
--- /dev/null
+++ b/tests/resources_plugin/test_github.py
@@ -0,0 +1,195 @@
+# 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.
+
+"""Test github resource plugin."""
+from unittest.mock import PropertyMock, patch
+
+import pytest
+
+from pydolphinscheduler.resources_plugin import GitHub
+from pydolphinscheduler.resources_plugin.base.git import GitFileInfo
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "file_path": "script/install.sh",
+ "api": "https://api.github.com/repos/{user}/{repo_name}/contents/{file_path}",
+ },
+ "https://api.github.com/repos/apache/dolphinscheduler/contents/script/install.sh",
+ ),
+ ],
+)
+def test_github_build_req_api(attr, expected):
+ """Test the build_req_api function of the github resource plug-in."""
+ github = GitHub(prefix="prefix")
+ assert expected == github.build_req_api(**attr)
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ "https://github.com/apache/dolphinscheduler/blob/dev/script/install.sh",
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "branch": "dev",
+ "file_path": "script/install.sh",
+ },
+ ),
+ (
+ "https://github.com/apache/dolphinscheduler/blob/master/pom.xml",
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "branch": "master",
+ "file_path": "pom.xml",
+ },
+ ),
+ (
+ "https://github.com/apache/dolphinscheduler/blob/1.3.9-release/docker/build/startup.sh",
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "branch": "1.3.9-release",
+ "file_path": "docker/build/startup.sh",
+ },
+ ),
+ ],
+)
+def test_github_get_git_file_info(attr, expected):
+ """Test the get_git_file_info function of the github resource plug-in."""
+ github = GitHub(prefix="prefix")
+ github.get_git_file_info(attr)
+ assert expected == github._git_file_info.__dict__
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ (
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "file_path": "docker/build/startup.sh",
+ }
+ ),
+ "https://api.github.com/repos/apache/dolphinscheduler/contents/docker/build/startup.sh",
+ ),
+ (
+ (
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "file_path": "pom.xml",
+ }
+ ),
+ "https://api.github.com/repos/apache/dolphinscheduler/contents/pom.xml",
+ ),
+ (
+ (
+ {
+ "user": "apache",
+ "repo_name": "dolphinscheduler",
+ "file_path": "script/create-dolphinscheduler.sh",
+ }
+ ),
+ "https://api.github.com/repos/apache/dolphinscheduler/contents/script/create-dolphinscheduler.sh",
+ ),
+ ],
+)
+@patch(
+ "pydolphinscheduler.resources_plugin.github.GitHub._git_file_info",
+ new_callable=PropertyMock,
+)
+def test_github_get_req_url(m_git_file_info, attr, expected):
+ """Test the get_req_url function of the github resource plug-in."""
+ github = GitHub(prefix="prefix")
+ m_git_file_info.return_value = GitFileInfo(**attr)
+ assert expected == github.get_req_url()
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "init": {"prefix": "prefix", "access_token": "access_token"},
+ "file_path": "github_resource_plugin.sh",
+ "file_content": "github resource plugin",
+ },
+ "github resource plugin",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "prefix",
+ },
+ "file_path": "github_resource_plugin.sh",
+ "file_content": "github resource plugin",
+ },
+ "github resource plugin",
+ ),
+ ],
+)
+@patch("pydolphinscheduler.resources_plugin.github.GitHub.req")
+def test_github_read_file(m_req, attr, expected):
+ """Test the read_file function of the github resource plug-in."""
+ github = GitHub(**attr.get("init"))
+ m_req.return_value = attr.get("file_content")
+ assert expected == github.read_file(attr.get("file_path"))
+
+
+@pytest.mark.skip(reason="Lack of test environment, need stable repository")
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ "https://github.com/apache/dolphinscheduler/blob/dev/lombok.config",
+ "#\n"
+ "# Licensed to the Apache Software Foundation (ASF) under one or more\n"
+ "# contributor license agreements. See the NOTICE file distributed with\n"
+ "# this work for additional information regarding copyright ownership.\n"
+ "# The ASF licenses this file to You under the Apache License, Version 2.0\n"
+ '# (the "License"); you may not use this file except in compliance with\n'
+ "# the License. You may obtain a copy of the License at\n"
+ "#\n"
+ "# http://www.apache.org/licenses/LICENSE-2.0\n"
+ "#\n"
+ "# Unless required by applicable law or agreed to in writing, software\n"
+ '# distributed under the License is distributed on an "AS IS" BASIS,\n'
+ "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n"
+ "# See the License for the specific language governing permissions and\n"
+ "# limitations under the License.\n"
+ "#\n"
+ "\n"
+ "lombok.addLombokGeneratedAnnotation = true\n",
+ ),
+ ],
+)
+def test_github_req(attr, expected):
+ """Test the req function of the github resource plug-in."""
+ github = GitHub(
+ prefix="prefix",
+ )
+ assert expected == github.req(attr)
diff --git a/tests/resources_plugin/test_gitlab.py b/tests/resources_plugin/test_gitlab.py
new file mode 100644
index 0000000..6bb90ac
--- /dev/null
+++ b/tests/resources_plugin/test_gitlab.py
@@ -0,0 +1,116 @@
+# 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.
+
+"""Test github resource plugin."""
+import pytest
+
+from pydolphinscheduler.resources_plugin.gitlab import GitLab
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ "https://gitlab.com/pydolphinscheduler/ds-gitlab/-/blob/main/union.sh",
+ {
+ "branch": "main",
+ "file_path": "union.sh",
+ "host": "https://gitlab.com",
+ "repo_name": "ds-gitlab",
+ "user": "pydolphinscheduler",
+ },
+ ),
+ (
+ "https://gitlab.com/pydolphinscheduler/ds/-/blob/dev/test/exc.sh",
+ {
+ "branch": "dev",
+ "file_path": "test/exc.sh",
+ "host": "https://gitlab.com",
+ "repo_name": "ds",
+ "user": "pydolphinscheduler",
+ },
+ ),
+ ],
+)
+def test_gitlab_get_git_file_info(attr, expected):
+ """Test the get_file_info function of the gitlab resource plugin."""
+ gitlab = GitLab(prefix="prefix")
+ gitlab.get_git_file_info(attr)
+ assert expected == gitlab._git_file_info.__dict__
+
+
+@pytest.mark.skip(reason="This test needs gitlab service")
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "init": {
+ "prefix": "https://gitlab.com/pydolphinscheduler/ds-internal/-/blob/main",
+ "oauth_token": "24518bd4cf5bfe9xx",
+ },
+ "file_path": "union.sh",
+ },
+ "test gitlab resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://gitlab.com/pydolphinscheduler/ds/-/blob/main",
+ "private_token": "9TyTe2xx",
+ },
+ "file_path": "union.sh",
+ },
+ "test gitlab resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://gitlab.com/pydolphinscheduler/ds-gitlab/-/blob/main",
+ "username": "pydolphinscheduler",
+ "password": "4295xx",
+ },
+ "file_path": "union.sh",
+ },
+ "test gitlab resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://gitlab.com/pydolphinscheduler/ds-public/-/blob/main",
+ },
+ "file_path": "union.sh",
+ },
+ "test gitlab resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://gitlab.com/pydolphinscheduler/ds-internal/-/blob/main",
+ "username": "pydolphinscheduler",
+ "password": "429xxx",
+ },
+ "file_path": "union.sh",
+ },
+ "test gitlab resource plugin\n",
+ ),
+ ],
+)
+def test_gitlab_read_file(attr, expected):
+ """Test the read_file function of the gitlab resource plug-in."""
+ gitlab = GitLab(**attr.get("init"))
+ assert expected == gitlab.read_file(attr.get("file_path"))
diff --git a/tests/resources_plugin/test_local.py b/tests/resources_plugin/test_local.py
new file mode 100644
index 0000000..82b196f
--- /dev/null
+++ b/tests/resources_plugin/test_local.py
@@ -0,0 +1,108 @@
+# 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.
+
+"""Test local resource plugin."""
+from pathlib import Path
+from unittest.mock import PropertyMock, patch
+
+import pytest
+
+from pydolphinscheduler.core import Task
+from pydolphinscheduler.exceptions import PyResPluginException
+from pydolphinscheduler.resources_plugin.local import Local
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file
+
+file_name = "local_res.sh"
+file_content = "echo Test local res plugin"
+res_plugin_prefix = Path(__file__).parent
+file_path = res_plugin_prefix.joinpath(file_name)
+
+
+@pytest.fixture()
+def setup_crt_first():
+ """Set up and teardown about create file first and then delete it."""
+ file.write(content=file_content, to_path=file_path)
+ yield
+ delete_file(file_path)
+
+
+@pytest.mark.parametrize(
+ "val, expected",
+ [
+ (file_name, file_content),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.ext",
+ new_callable=PropertyMock,
+ return_value={
+ ".sh",
+ },
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.ext_attr",
+ new_callable=PropertyMock,
+ return_value="_raw_script",
+)
+@patch(
+ "pydolphinscheduler.core.task.Task._raw_script",
+ create=True,
+ new_callable=PropertyMock,
+)
+def test_task_obtain_res_plugin(
+ m_raw_script, m_ext_attr, m_ext, m_code_version, val, expected, setup_crt_first
+):
+ """Test task obtaining resource plug-in."""
+ m_raw_script.return_value = val
+ task = Task(
+ name="test_task_ext_attr",
+ task_type="type",
+ resource_plugin=Local(str(res_plugin_prefix)),
+ )
+ assert expected == getattr(task, "raw_script")
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [({"prefix": res_plugin_prefix, "file_name": file_name}, file_content)],
+)
+def test_local_res_read_file(attr, expected, setup_crt_first):
+ """Test the read_file function of the local resource plug-in."""
+ local = Local(str(attr.get("prefix")))
+ local.read_file(attr.get("file_name"))
+ assert expected == local.read_file(file_name)
+
+
+@pytest.mark.parametrize(
+ "attr",
+ [
+ {"prefix": res_plugin_prefix, "file_name": file_name},
+ ],
+)
+def test_local_res_file_not_found(attr):
+ """Test local resource plugin file does not exist."""
+ with pytest.raises(
+ PyResPluginException,
+ match=".* is not found",
+ ):
+ local = Local(str(attr.get("prefix")))
+ local.read_file(attr.get("file_name"))
diff --git a/tests/resources_plugin/test_oss.py b/tests/resources_plugin/test_oss.py
new file mode 100644
index 0000000..7e57e82
--- /dev/null
+++ b/tests/resources_plugin/test_oss.py
@@ -0,0 +1,112 @@
+# 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.
+
+"""Test oss resource plugin."""
+import pytest
+
+from pydolphinscheduler.resources_plugin.oss import OSS
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ "https://ospp-ds-private.oss-cn-hangzhou.aliyuncs.com/a.sh",
+ {
+ "endpoint": "https://oss-cn-hangzhou.aliyuncs.com",
+ "file_path": "a.sh",
+ "bucket": "ospp-ds-private",
+ },
+ ),
+ (
+ "https://ospp-ds-public.oss-cn-hangzhou.aliyuncs.com/dir/a.sh",
+ {
+ "endpoint": "https://oss-cn-hangzhou.aliyuncs.com",
+ "file_path": "dir/a.sh",
+ "bucket": "ospp-ds-public",
+ },
+ ),
+ ],
+)
+def test_oss_get_bucket_file_info(attr, expected):
+ """Test the get_bucket_file_info function of the oss resource plugin."""
+ oss = OSS(prefix="prefix")
+ oss.get_bucket_file_info(attr)
+ assert expected == oss._bucket_file_info.__dict__
+
+
+@pytest.mark.skip(reason="This test requires OSS services")
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "init": {
+ "prefix": "https://ospp-ds-private.oss-cn-hangzhou.aliyuncs.com",
+ "access_key_id": "LTAI5tP25Mxx",
+ "access_key_secret": "cSur23Qbxx",
+ },
+ "file_path": "a.sh",
+ },
+ "test oss resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://ospp-ds-private.oss-cn-hangzhou.aliyuncs.com/dir/",
+ "access_key_id": "LTAxx",
+ "access_key_secret": "cSur23Qxx",
+ },
+ "file_path": "b.sh",
+ },
+ "test oss resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://ospp-ds-private.oss-cn-hangzhou.aliyuncs.com",
+ },
+ "file_path": "b.sh",
+ },
+ "test oss resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://ospp-ds-public.oss-cn-hangzhou.aliyuncs.com",
+ },
+ "file_path": "b.sh",
+ },
+ "test oss resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://ospp-ds-public.oss-cn-hangzhou.aliyuncs.com/dir/",
+ "access_key_id": "LTAIxx",
+ "access_key_secret": "cSurxx",
+ },
+ "file_path": "a.sh",
+ },
+ "test oss resource plugin\n",
+ ),
+ ],
+)
+def test_oss_read_file(attr, expected):
+ """Test the read_file function of the oss resource plug-in."""
+ oss = OSS(**attr.get("init"))
+ assert expected == oss.read_file(attr.get("file_path"))
diff --git a/tests/resources_plugin/test_resource_plugin.py b/tests/resources_plugin/test_resource_plugin.py
new file mode 100644
index 0000000..63e619a
--- /dev/null
+++ b/tests/resources_plugin/test_resource_plugin.py
@@ -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.
+
+"""Test abstract class resource_plugin."""
+
+import pytest
+
+from pydolphinscheduler.exceptions import PyResPluginException
+from pydolphinscheduler.resources_plugin import GitHub
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "s": "https://api.github.com/repos/apache/dolphinscheduler/contents/script/install.sh",
+ "x": "/",
+ "n": 2,
+ },
+ 7,
+ ),
+ (
+ {
+ "s": "https://api.github.com",
+ "x": ":",
+ "n": 1,
+ },
+ 5,
+ ),
+ ],
+)
+def test_github_get_index(attr, expected):
+ """Test the get_index function of the abstract class resource_plugin."""
+ github = GitHub(prefix="prefix")
+ assert expected == github.get_index(**attr)
+
+
+@pytest.mark.parametrize(
+ "attr",
+ [
+ {
+ "s": "https://api.github.com",
+ "x": "/",
+ "n": 3,
+ },
+ {
+ "s": "https://api.github.com/",
+ "x": "/",
+ "n": 4,
+ },
+ ],
+)
+def test_github_get_index_exception(attr):
+ """Test exception to get_index function of abstract class resource_plugin."""
+ with pytest.raises(
+ PyResPluginException,
+ match="Incomplete path.",
+ ):
+ github = GitHub(prefix="prefix")
+ github.get_index(**attr)
diff --git a/tests/resources_plugin/test_s3.py b/tests/resources_plugin/test_s3.py
new file mode 100644
index 0000000..5f75f3e
--- /dev/null
+++ b/tests/resources_plugin/test_s3.py
@@ -0,0 +1,79 @@
+# 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.
+
+"""Test oss resource plugin."""
+import pytest
+
+from pydolphinscheduler.resources_plugin import S3
+
+
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ "https://ds-resource-plugin-private.s3.amazonaws.com/a.sh",
+ {
+ "file_path": "a.sh",
+ "bucket": "ds-resource-plugin-private",
+ },
+ ),
+ (
+ "https://ds-resource-plugin-public.s3.amazonaws.com/dir/a.sh",
+ {
+ "file_path": "dir/a.sh",
+ "bucket": "ds-resource-plugin-public",
+ },
+ ),
+ ],
+)
+def test_s3_get_bucket_file_info(attr, expected):
+ """Test the get_bucket_file_info function of the s3 resource plugin."""
+ s3 = S3(prefix="prefix")
+ s3.get_bucket_file_info(attr)
+ assert expected == s3._bucket_file_info.__dict__
+
+
+@pytest.mark.skip(reason="This test requires s3 services")
+@pytest.mark.parametrize(
+ "attr, expected",
+ [
+ (
+ {
+ "init": {
+ "prefix": "https://ds-resource-plugin-private.s3.amazonaws.com/dir/",
+ "access_key_id": "LTAI5tP25Mxx",
+ "access_key_secret": "cSur23Qbxx",
+ },
+ "file_path": "a.sh",
+ },
+ "test s3 resource plugin\n",
+ ),
+ (
+ {
+ "init": {
+ "prefix": "https://ds-resource-plugin-public.s3.amazonaws.com/",
+ },
+ "file_path": "a.sh",
+ },
+ "test s3 resource plugin\n",
+ ),
+ ],
+)
+def test_s3_read_file(attr, expected):
+ """Test the read_file function of the s3 resource plug-in."""
+ s3 = S3(**attr.get("init"))
+ assert expected == s3.read_file(attr.get("file_path"))
diff --git a/tests/tasks/test_datax.py b/tests/tasks/test_datax.py
index 5d1890e..95f65b3 100644
--- a/tests/tasks/test_datax.py
+++ b/tests/tasks/test_datax.py
@@ -16,12 +16,28 @@
# under the License.
"""Test Task DataX."""
-
+from pathlib import Path
from unittest.mock import patch
import pytest
+from pydolphinscheduler.resources_plugin import Local
from pydolphinscheduler.tasks.datax import CustomDataX, DataX
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file
+
+
+@pytest.fixture()
+def setup_crt_first(request):
+ """Set up and teardown about create file first and then delete it."""
+ file_content = request.param.get("file_content")
+ file_path = request.param.get("file_path")
+ file.write(
+ content=file_content,
+ to_path=file_path,
+ )
+ yield
+ delete_file(file_path)
@patch(
@@ -122,3 +138,76 @@ def test_custom_datax_get_define(json_template):
):
task = CustomDataX(name, json_template)
assert task.get_define() == expect
+
+
+@pytest.mark.parametrize(
+ "setup_crt_first",
+ [
+ {
+ "file_path": Path(__file__).parent.joinpath("local_res.sql"),
+ "file_content": "test local resource",
+ }
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize(
+ "attr, expect",
+ [
+ (
+ {
+ "name": "task_datax",
+ "datasource_name": "first_mysql",
+ "datatarget_name": "second_mysql",
+ "sql": "local_res.sql",
+ "target_table": "target_table",
+ "resource_plugin": Local(str(Path(__file__).parent)),
+ },
+ "test local resource",
+ ),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+def test_resources_local_datax_command_content(
+ mock_code_version, attr, expect, setup_crt_first
+):
+ """Test task datax sql content through the local resource plug-in."""
+ datax = DataX(**attr)
+ assert expect == getattr(datax, "sql")
+
+
+@pytest.mark.parametrize(
+ "setup_crt_first",
+ [
+ {
+ "file_path": Path(__file__).parent.joinpath("local_res.json"),
+ "file_content": '{content: "test local resource"}',
+ }
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize(
+ "attr, expect",
+ [
+ (
+ {
+ "name": "task_custom_datax",
+ "json": "local_res.json",
+ "resource_plugin": Local(str(Path(__file__).parent)),
+ },
+ '{content: "test local resource"}',
+ ),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+def test_resources_local_custom_datax_command_content(
+ mock_code_version, attr, expect, setup_crt_first
+):
+ """Test task CustomDataX json content through the local resource plug-in."""
+ custom_datax = CustomDataX(**attr)
+ assert expect == getattr(custom_datax, "json")
diff --git a/tests/tasks/test_python.py b/tests/tasks/test_python.py
index e8f7f10..77aa106 100644
--- a/tests/tasks/test_python.py
+++ b/tests/tasks/test_python.py
@@ -16,26 +16,42 @@
# under the License.
"""Test Task python."""
-
-
+from pathlib import Path
from unittest.mock import patch
import pytest
from pydolphinscheduler.exceptions import PyDSParamException
+from pydolphinscheduler.resources_plugin import Local
from pydolphinscheduler.tasks.python import Python
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file
def foo(): # noqa: D103
print("hello world.")
+@pytest.fixture()
+def setup_crt_first(request):
+ """Set up and teardown about create file first and then delete it."""
+ file_content = request.param.get("file_content")
+ file_path = request.param.get("file_path")
+ file.write(
+ content=file_content,
+ to_path=file_path,
+ )
+ yield
+ delete_file(file_path)
+
+
@pytest.mark.parametrize(
"attr, expect",
[
(
{"definition": "print(1)"},
{
+ "definition": "print(1)",
"rawScript": "print(1)",
"localParams": [],
"resourceList": [],
@@ -47,6 +63,7 @@ def foo(): # noqa: D103
(
{"definition": "def foo():\n print('I am foo')"},
{
+ "definition": "def foo():\n print('I am foo')",
"rawScript": "def foo():\n print('I am foo')\nfoo()",
"localParams": [],
"resourceList": [],
@@ -58,6 +75,7 @@ def foo(): # noqa: D103
(
{"definition": foo},
{
+ "definition": foo,
"rawScript": 'def foo(): # noqa: D103\n print("hello world.")\nfoo()',
"localParams": [],
"resourceList": [],
@@ -122,6 +140,7 @@ def test_python_get_define(name, script_code, raw):
"delayTime": 0,
"taskType": "PYTHON",
"taskParams": {
+ "definition": script_code,
"resourceList": [],
"localParams": [],
"rawScript": raw,
@@ -145,3 +164,38 @@ def test_python_get_define(name, script_code, raw):
):
shell = Python(name, script_code)
assert shell.get_define() == expect
+
+
+@pytest.mark.parametrize(
+ "setup_crt_first",
+ [
+ {
+ "file_path": Path(__file__).parent.joinpath("local_res.py"),
+ "file_content": "test local resource",
+ }
+ ],
+ indirect=True,
+)
+@pytest.mark.parametrize(
+ "attr, expect",
+ [
+ (
+ {
+ "name": "task_python",
+ "definition": "local_res.py",
+ "resource_plugin": Local(str(Path(__file__).parent)),
+ },
+ "test local resource",
+ ),
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+def test_resources_local_python_command_content(
+ mock_code_version, attr, expect, setup_crt_first
+):
+ """Test task Python definition content through the local resource plug-in."""
+ python = Python(**attr)
+ assert expect == getattr(python, "definition")
diff --git a/tests/tasks/test_shell.py b/tests/tasks/test_shell.py
index e2c87d8..9344ac2 100644
--- a/tests/tasks/test_shell.py
+++ b/tests/tasks/test_shell.py
@@ -17,12 +17,28 @@
"""Test Task shell."""
-
+from pathlib import Path
from unittest.mock import patch
import pytest
+from pydolphinscheduler.resources_plugin import Local
from pydolphinscheduler.tasks.shell import Shell
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file
+
+file_name = "local_res.sh"
+file_content = 'echo "test res_local"'
+res_plugin_prefix = Path(__file__).parent
+file_path = res_plugin_prefix.joinpath(file_name)
+
+
+@pytest.fixture
+def setup_crt_first():
+ """Set up and teardown about create file first and then delete it."""
+ file.write(content=file_content, to_path=file_path)
+ yield
+ delete_file(file_path)
@pytest.mark.parametrize(
@@ -90,3 +106,28 @@ def test_shell_get_define():
shell = Shell(name, command)
print(shell.get_define())
assert shell.get_define() == expect
+
+
+@pytest.mark.parametrize(
+ "attr, expect",
+ [
+ (
+ {
+ "name": "test-local-res-command-content",
+ "command": file_name,
+ "resource_plugin": Local(str(res_plugin_prefix)),
+ },
+ file_content,
+ )
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+def test_resources_local_shell_command_content(
+ mock_code_version, attr, expect, setup_crt_first
+):
+ """Test task shell task command content through the local resource plug-in."""
+ task = Shell(**attr)
+ assert expect == getattr(task, "raw_script")
diff --git a/tests/tasks/test_spark.py b/tests/tasks/test_spark.py
index ed83f9f..1fdb1fa 100644
--- a/tests/tasks/test_spark.py
+++ b/tests/tasks/test_spark.py
@@ -19,7 +19,7 @@
from unittest.mock import patch
-from pydolphinscheduler.tasks.spark import DeployMode, ProgramType, Spark, SparkVersion
+from pydolphinscheduler.tasks.spark import DeployMode, ProgramType, Spark
@patch(
@@ -50,7 +50,6 @@ def test_spark_get_define(mock_resource):
},
"programType": program_type,
"deployMode": deploy_mode,
- "sparkVersion": SparkVersion.SPARK2,
"driverCores": 1,
"driverMemory": "512M",
"numExecutors": 2,
diff --git a/tests/tasks/test_sql.py b/tests/tasks/test_sql.py
index ba9daa9..a22d920 100644
--- a/tests/tasks/test_sql.py
+++ b/tests/tasks/test_sql.py
@@ -16,13 +16,28 @@
# under the License.
"""Test Task Sql."""
-
-
+from pathlib import Path
from unittest.mock import patch
import pytest
+from pydolphinscheduler.resources_plugin import Local
from pydolphinscheduler.tasks.sql import Sql, SqlType
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file
+
+file_name = "local_res.sql"
+file_content = "select 1"
+res_plugin_prefix = Path(__file__).parent
+file_path = res_plugin_prefix.joinpath(file_name)
+
+
+@pytest.fixture
+def setup_crt_first():
+ """Set up and teardown about create file first and then delete it."""
+ file.write(content=file_content, to_path=file_path)
+ yield
+ delete_file(file_path)
@pytest.mark.parametrize(
@@ -165,3 +180,29 @@ def test_sql_get_define(mock_datasource):
):
task = Sql(name, datasource_name, command)
assert task.get_define() == expect
+
+
+@pytest.mark.parametrize(
+ "attr, expect",
+ [
+ (
+ {
+ "name": "test-sql-local-res",
+ "sql": file_name,
+ "datasource_name": "test_datasource",
+ "resource_plugin": Local(str(res_plugin_prefix)),
+ },
+ file_content,
+ )
+ ],
+)
+@patch(
+ "pydolphinscheduler.core.task.Task.gen_code_and_version",
+ return_value=(123, 1),
+)
+def test_resources_local_sql_command_content(
+ mock_code_version, attr, expect, setup_crt_first
+):
+ """Test sql content through the local resource plug-in."""
+ sql = Sql(**attr)
+ assert expect == getattr(sql, "sql")
diff --git a/tests/test_java_gateway.py b/tests/test_java_gateway.py
deleted file mode 100644
index 3c8831e..0000000
--- a/tests/test_java_gateway.py
+++ /dev/null
@@ -1,52 +0,0 @@
-# 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.
-
-"""Test pydolphinscheduler java gateway."""
-
-
-from py4j.java_gateway import JavaGateway, java_import
-
-
-def test_gateway_connect():
- """Test weather client could connect java gate way or not."""
- gateway = JavaGateway()
- app = gateway.entry_point
- assert app.ping() == "PONG"
-
-
-def test_jvm_simple():
- """Test use JVM build-in object and operator from java gateway."""
- gateway = JavaGateway()
- smaller = gateway.jvm.java.lang.Integer.MIN_VALUE
- bigger = gateway.jvm.java.lang.Integer.MAX_VALUE
- assert bigger > smaller
-
-
-def test_python_client_java_import_single():
- """Test import single class from java gateway."""
- gateway = JavaGateway()
- java_import(gateway.jvm, "org.apache.dolphinscheduler.common.utils.FileUtils")
- assert hasattr(gateway.jvm, "FileUtils")
-
-
-def test_python_client_java_import_package():
- """Test import package contain multiple class from java gateway."""
- gateway = JavaGateway()
- java_import(gateway.jvm, "org.apache.dolphinscheduler.common.utils.*")
- # test if jvm view have some common utils
- for util in ("FileUtils", "OSUtils", "DateUtils"):
- assert hasattr(gateway.jvm, util)