You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/07/03 14:19:32 UTC
[GitHub] [airflow] AngryHelper opened a new pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
AngryHelper opened a new pull request #16792:
URL: https://github.com/apache/airflow/pull/16792
closes: AIRFLOW-5703
Added next features: transfer list of files, transfer directory, transfer files on regex pattern in directories.
<details>
<summary>I writen next dag for tests</summary>
```
from airflow.utils.edgemodifier import Label
from airflow.operators.bash import BashOperator
from airflow import DAG
from airflow.providers.sftp.operators.sftp import SFTPOperator, SFTPOperation
from airflow.providers.ssh.operators.ssh import SSHOperator
from airflow.utils.dates import days_ago
with DAG(
dag_id="test_stfp",
schedule_interval="@once",
start_date=days_ago(1),
max_active_runs=1
) as dag:
create_test_files = BashOperator(
task_id="create_test_files",
bash_command="""
touch /tmp/file.txt & echo "transfer data from local" >> /tmp/file.txt &
touch /tmp/file1.txt & echo "transfer data from local" >> /tmp/file1.txt &
touch /tmp/file2.txt & echo "transfer data from local" >> /tmp/file2.txt &
touch /tmp/file3.txt & echo "transfer data from local" >> /tmp/file3.txt &
mkdir /tmp/dir_for_remote_transfer &
touch /tmp/dir_for_remote_transfer/file2.txt &
touch /tmp/dir_for_remote_transfer/file3.csv &
ls /tmp """
)
put_file = SFTPOperator(
task_id="put_file",
ssh_conn_id="ssh_default",
local_filepath="/tmp/file.txt",
remote_filepath="/tmp/tmp1/tmp2/file.txt",
operation=SFTPOperation.PUT,
create_intermediate_dirs=True
)
put_file_without_full_remote_path = SFTPOperator(
task_id="put_file_without_full_remote_path",
ssh_conn_id="ssh_default",
local_filepath="/tmp/file3.txt",
remote_filepath="/tmp/tmp1/tmp2/",
operation=SFTPOperation.PUT,
create_intermediate_dirs=True
)
put_dir_txt_files = SFTPOperator(
task_id="put_dir_txt_files",
ssh_conn_id="ssh_default",
local_folder="/tmp/dir_for_remote_transfer/",
remote_folder="/tmp/dir_for_remote_transfer/txt",
regexp_mask=".*.txt",
operation=SFTPOperation.PUT,
create_intermediate_dirs=True
)
put_dir_csv_files = SFTPOperator(
task_id="put_dir_csv_files",
ssh_conn_id="ssh_default",
local_folder="/tmp/dir_for_remote_transfer/",
remote_folder="/tmp/dir_for_remote_transfer/csv/",
regexp_mask=".*.csv",
operation=SFTPOperation.PUT,
create_intermediate_dirs=True
)
put_files = SFTPOperator(
task_id="put_files",
ssh_conn_id="ssh_default",
local_filepath=["/tmp/file1.txt", "/tmp/file2.txt"],
remote_filepath="/tmp/tmp1/tmp2/",
operation=SFTPOperation.PUT,
create_intermediate_dirs=True
)
drop_local_test_files = BashOperator(
task_id="drop_local_test_files",
bash_command="""rm -rf /tmp/file* & rm -rf /tmp/dir_for_remote_transfer/ & ls /tmp/"""
)
get_file = SFTPOperator(
task_id="get_file",
ssh_conn_id="ssh_default",
local_filepath="/tmp/file.txt",
remote_filepath="/tmp/tmp1/tmp2/file.txt",
operation=SFTPOperation.GET,
create_intermediate_dirs=True
)
get_files = SFTPOperator(
task_id="get_files",
ssh_conn_id="ssh_default",
local_filepath="/tmp/",
remote_filepath=["/tmp/tmp1/tmp2/file1.txt", "/tmp/tmp1/tmp2/file2.txt"],
operation=SFTPOperation.GET,
create_intermediate_dirs=True
)
get_dir = SFTPOperator(
task_id="get_dir",
ssh_conn_id="ssh_default",
local_folder="/tmp/dir_for_remote_transfer/csv",
remote_folder="/tmp/dir_for_remote_transfer/csv",
operation=SFTPOperation.GET,
create_intermediate_dirs=True
)
drop_remote_files = SSHOperator(
task_id="drop_remote_files",
ssh_conn_id="ssh_default",
command="rm -rf /tmp/tmp1 & ls /tmp/"
)
create_test_files >> Label("transfer data to remote server") >> \
[put_file, put_files, put_file_without_full_remote_path, put_dir_txt_files, put_dir_csv_files] >> \
drop_local_test_files >> Label("transfer data to local server") >> \
[get_file, get_files, get_dir] >> drop_remote_files
```
</details>
<details>
<summary>Pre-commit Tests</summary>
```
ubuntu@ubuntu-advanced-8-16-10gb:~/airflow$ pre-commit run --files airflow/providers/sftp/operators/sftp.py
identity.................................................................................Passed
- hook id: identity
- duration: 0.04s
airflow/providers/sftp/operators/sftp.py
Check hooks apply to the repository..................................(no files to check)Skipped
No-tabs checker..........................................................................Passed
Add license for all SQL files........................................(no files to check)Skipped
Add license for all other files......................................(no files to check)Skipped
Add license for all rst files........................................(no files to check)Skipped
Add license for all JS/TS/TSX/CSS/PUML files.........................(no files to check)Skipped
Add license for all JINJA template files.............................(no files to check)Skipped
Add license for all shell files......................................(no files to check)Skipped
Add license for all Python files.........................................................Passed
Add license for all XML files........................................(no files to check)Skipped
Add license for all YAML files.......................................(no files to check)Skipped
Add license for all md files.........................................(no files to check)Skipped
Add license for all mermaid files....................................(no files to check)Skipped
Add TOC for md files.................................................(no files to check)Skipped
black....................................................................................Passed
blacken-docs.............................................................................Passed
Check for merge conflicts................................................................Passed
Debug Statements (Python)................................................................Passed
Check builtin type constructor use.......................................................Passed
Detect Private Key.......................................................................Passed
Fix End of Files.........................................................................Passed
Mixed line ending........................................................................Passed
Check that executables have shebangs.................................(no files to check)Skipped
Check Xml............................................................(no files to check)Skipped
Trim Trailing Whitespace.................................................................Passed
Fix python encoding pragma...............................................................Passed
Pretty format JSON...................................................(no files to check)Skipped
pyupgrade................................................................................Passed
rst ``code`` is two backticks........................................(no files to check)Skipped
use logger.warning(......................................................................Passed
Check YAML files with yamllint.......................................(no files to check)Skipped
Run isort to sort imports in Python files................................................Passed
Run pydocstyle...........................................................................Passed
Strip unnecessary `# noqa`s..............................................................Passed
Lint OpenAPI using spectral..........................................(no files to check)Skipped
Lint OpenAPI using openapi-spec-validator............................(no files to check)Skipped
Lint dockerfile......................................................(no files to check)Skipped
Check order of dependencies in setup.cfg and setup.py................(no files to check)Skipped
Checks setup extra packages..........................................(no files to check)Skipped
Update output of breeze command in BREEZE.rst........................(no files to check)Skipped
Update mounts in the local yml file..................................(no files to check)Skipped
Update setup.cfg file with all licenses..............................(no files to check)Skipped
Build cross-dependencies for providers packages..........................................Passed
Update extras in documentation.......................................(no files to check)Skipped
Check for pydevd debug statements accidentally left......................................Passed
Don't use safe in templates..........................................(no files to check)Skipped
No providers imports in core example DAGs............................(no files to check)Skipped
No relative imports......................................................................Passed
Check for language that we do not accept as community....................................Passed
Check BaseOperator[Link] core imports................................(no files to check)Skipped
Check BaseOperator[Link] other imports...................................................Passed
Check provide_session and create_session imports.........................................Passed
Make sure LoggingMixin is not used alone.................................................Passed
Make sure days_ago is imported from airflow.utils.dates..................................Passed
'start_date' should not be defined in default_args in example_dags...(no files to check)Skipped
Check if integration list is aligned.................................(no files to check)Skipped
Check if licenses are OK for Apache..................................(no files to check)Skipped
Checks for consistency between config.yml and default_config.cfg.....(no files to check)Skipped
Verify that DB Migrations have been documented.......................(no files to check)Skipped
Checks for Boring Cyborg configuration consistency.......................................Passed
Sort INTHEWILD.md alphabetically.....................................(no files to check)Skipped
Sort alphabetically and uniquify spelling_wordlist.txt...............(no files to check)Skipped
Lint Helm Chart......................................................(no files to check)Skipped
Check Shell scripts syntax correctness...............................(no files to check)Skipped
Run BATS bash tests for changed Breeze bash files....................(no files to check)Skipped
Run BATS bash tests for changed bash files...........................(no files to check)Skipped
stylelint............................................................(no files to check)Skipped
Provider init file.......................................................................Passed
Validate providers.yaml files........................................(no files to check)Skipped
Generate mermaid images..............................................(no files to check)Skipped
Check if pre-commits are described...................................(no files to check)Skipped
Ensure hook ids are not overly long..................................(no files to check)Skipped
Checks providers available when declared by extras in setup.py...........................Passed
Run markdownlint.....................................................(no files to check)Skipped
Lint JSON Schema files with JSON Schema..............................(no files to check)Skipped
Lint NodePort Service with JSON Schema...............................(no files to check)Skipped
Lint Docker compose files with JSON Schema...........................(no files to check)Skipped
Lint chart/values.schema.json file with JSON Schema..................(no files to check)Skipped
Lint chart/values.yaml file with JSON Schema.........................(no files to check)Skipped
Lint airflow/config_templates/config.yml file with JSON Schema.......(no files to check)Skipped
Convert to f-strings with flynt..........................................................Passed
ESLint against airflow/ui............................................(no files to check)Skipped
ESLint against current UI js files...................................(no files to check)Skipped
Check if image build is needed...........................................................Passed
Run in container bats tests..........................................(no files to check)Skipped
Run mypy.................................................................................Passed
Run mypy for helm chart tests........................................(no files to check)Skipped
Run mypy for /docs/ folder...........................................(no files to check)Skipped
Run flake8...............................................................................Passed
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r663497737
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
Why `ntpath` specifically, and only this function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-908647425
sorry for flood, I want create another PR with true commits
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r671712201
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
You have mistake in your second comparision (missing "txt" in second example) - that's why you have difference
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r663529864
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
I needed to take the name of file from full path, this function was just right for this task :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-907842770
> Need to fix linter issues. See guide: https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#contribution-workflow
sorry for your waiting, I was solving the issue with my cloud provider.
I pushed fix about linter issues, you may see and pls run workflow, as I understand, I don`t have access for this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-908509451
Can you please rebase to latest `main` @AngryHelper ? This will fix the failing image build,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-873417015
> Also the DAG you wrote for tests is great to have an example dag in documentation. So I think it should be added too as a really nice documentation similarly as we have in other providers. It should not be difficult to add them.
>
> You can take as an example the airbyte provider:
>
> Example dag here for airbyte:
>
> https://github.com/apache/airflow/tree/main/airflow/providers/airbyte/example_dags
>
> And parts of the example dag are included in airbyte documentation:
>
> https://github.com/apache/airflow/blob/main/docs/apache-airflow-providers-airbyte/operators/airbyte.rst
>
> This way your example dag will become a guide for others :)
thanks for the examples, I will definitely improve it!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r671713104
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
you right, tnx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r698291886
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,81 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
- else:
- remote_folder = os.path.dirname(self.remote_filepath)
- if self.create_intermediate_dirs:
- _make_intermediate_dirs(
- sftp_client=sftp_client,
- remote_directory=remote_folder,
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ local_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, local_folder, local_file, self.remote_filepath)
+ elif isinstance(self.remote_filepath, list) and isinstance(self.local_filepath, str):
+ for file_path in self.remote_filepath:
+ remote_folder = os.path.dirname(file_path)
+ remote_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, self.local_filepath, remote_file, remote_folder)
+ elif isinstance(self.remote_filepath, str) and isinstance(self.local_filepath, str):
+ local_folder = os.path.dirname(self.local_filepath)
+ file_msg = self.local_filepath
+ self._transfer(
+ sftp_client,
+ local_folder,
+ self.local_filepath,
+ self.remote_filepath,
+ only_file=True,
)
- file_msg = f"from {self.local_filepath} to {self.remote_filepath}"
- self.log.info("Starting to transfer file %s", file_msg)
- sftp_client.put(self.local_filepath, self.remote_filepath, confirm=self.confirm)
+ elif self.local_folder and self.remote_folder:
+ if self.operation.lower() == SFTPOperation.PUT:
+ files_list = self._search_files(os.listdir(self.local_folder))
+ for file in files_list:
+ local_file = os.path.basename(file)
+ file_msg = file
+ self._transfer(sftp_client, self.local_folder, local_file, self.remote_folder)
+ elif self.operation.lower() == SFTPOperation.GET:
+ files_list = self._search_files(sftp_client.listdir(self.remote_folder))
+ for file in files_list:
+ remote_file = ntpath.basename(file)
Review comment:
I think this should be `os.path.basename` like in the `PUT` block.
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -88,6 +124,9 @@ def __init__(
remote_host=None,
local_filepath=None,
remote_filepath=None,
+ local_folder=None,
+ remote_folder=None,
+ regexp_mask=None,
Review comment:
I am really not fond of this argument list since it’s too easy to pass in nonsensical combinations e.g.
```python
SFTPOperator(
local_filepath="/my/file",
local_folder="/my/dir",
)
```
But I understand this is not easy to change the arguments due to compatibility considerations.
Perhapes what we should do is to introduce a separate operator (say `SFTPBatchOperator`) to handle the new feature, instead of bolting it onto the existing one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper edited a comment on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper edited a comment on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-907842770
> Need to fix linter issues. See guide: https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#contribution-workflow
@uranusjr @potiuk sorry for your waiting, I was solving the issue with my cloud provider.
I pushed fix about linter issues, you may see and pls run workflow, as I understand, I don`t have access for this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r698559013
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -88,6 +124,9 @@ def __init__(
remote_host=None,
local_filepath=None,
remote_filepath=None,
+ local_folder=None,
+ remote_folder=None,
+ regexp_mask=None,
Review comment:
for this case will print AirflowException with examples
https://github.com/apache/airflow/pull/16792/files#diff-911f39c07d851bc2f0a2958674ed940ddc68aea6ecc7d71c29567816c182eaf6R216
I checked the example of dag again, I am comfortable working with one operetor, even if he has combined arguments.
Your idea i understand too, may be we will wait @potiuk and hear his opinion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r671712201
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
You have mistake in your second comparision (missing "txt" in third example) - that's why you have difference
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r671711913
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
its difference between functions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-873416594
Also the DAG you wrote for tests is great to have an example dag in documentation. So I think it should be added too as a really nice documentation similarly as we have in other providers. It should not be difficult to add them.
You can take as an example the airbyte provider:
Example dag here for airbyte:
https://github.com/apache/airflow/tree/main/airflow/providers/airbyte/example_dags
And parts of the example dag are included in airbyte documentation:
https://github.com/apache/airflow/blob/main/docs/apache-airflow-providers-airbyte/operators/airbyte.rst
This way your example dag will become a guide for others :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-908509451
Can you please rebase to latest `main` @AngryHelper ? This will fix the failing image build,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r698591513
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,81 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
- else:
- remote_folder = os.path.dirname(self.remote_filepath)
- if self.create_intermediate_dirs:
- _make_intermediate_dirs(
- sftp_client=sftp_client,
- remote_directory=remote_folder,
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ local_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, local_folder, local_file, self.remote_filepath)
+ elif isinstance(self.remote_filepath, list) and isinstance(self.local_filepath, str):
+ for file_path in self.remote_filepath:
+ remote_folder = os.path.dirname(file_path)
+ remote_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, self.local_filepath, remote_file, remote_folder)
+ elif isinstance(self.remote_filepath, str) and isinstance(self.local_filepath, str):
+ local_folder = os.path.dirname(self.local_filepath)
+ file_msg = self.local_filepath
+ self._transfer(
+ sftp_client,
+ local_folder,
+ self.local_filepath,
+ self.remote_filepath,
+ only_file=True,
)
- file_msg = f"from {self.local_filepath} to {self.remote_filepath}"
- self.log.info("Starting to transfer file %s", file_msg)
- sftp_client.put(self.local_filepath, self.remote_filepath, confirm=self.confirm)
+ elif self.local_folder and self.remote_folder:
+ if self.operation.lower() == SFTPOperation.PUT:
+ files_list = self._search_files(os.listdir(self.local_folder))
+ for file in files_list:
+ local_file = os.path.basename(file)
+ file_msg = file
+ self._transfer(sftp_client, self.local_folder, local_file, self.remote_folder)
+ elif self.operation.lower() == SFTPOperation.GET:
+ files_list = self._search_files(sftp_client.listdir(self.remote_folder))
+ for file in files_list:
+ remote_file = ntpath.basename(file)
Review comment:
yep, sorry, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r663567017
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,68 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ remote_folder = os.path.dirname(self.remote_filepath)
+ local_file = ntpath.basename(file_path)
Review comment:
But why not `os.path.basename`? `ntpath` splits on backslashes (as it should) and that would cause issues on Linux (where ``\`` is a valid character in file name).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on a change in pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on a change in pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#discussion_r698559013
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -88,6 +124,9 @@ def __init__(
remote_host=None,
local_filepath=None,
remote_filepath=None,
+ local_folder=None,
+ remote_folder=None,
+ regexp_mask=None,
Review comment:
for this case will print AirflowException with examples
https://github.com/apache/airflow/pull/16792/files#diff-911f39c07d851bc2f0a2958674ed940ddc68aea6ecc7d71c29567816c182eaf6R216
I checked the example of dag again, I am comfortable working with one operetor, even if he has combined arguments.
Your idea i understand too, may be we will wait @potiuk and hear his opinion
##########
File path: airflow/providers/sftp/operators/sftp.py
##########
@@ -134,29 +176,81 @@ def execute(self, context: Any) -> str:
with self.ssh_hook.get_conn() as ssh_client:
sftp_client = ssh_client.open_sftp()
- if self.operation.lower() == SFTPOperation.GET:
- local_folder = os.path.dirname(self.local_filepath)
- if self.create_intermediate_dirs:
- Path(local_folder).mkdir(parents=True, exist_ok=True)
- file_msg = f"from {self.remote_filepath} to {self.local_filepath}"
- self.log.info("Starting to transfer %s", file_msg)
- sftp_client.get(self.remote_filepath, self.local_filepath)
- else:
- remote_folder = os.path.dirname(self.remote_filepath)
- if self.create_intermediate_dirs:
- _make_intermediate_dirs(
- sftp_client=sftp_client,
- remote_directory=remote_folder,
+ if self.local_filepath and self.remote_filepath:
+ if isinstance(self.local_filepath, list) and isinstance(self.remote_filepath, str):
+ for file_path in self.local_filepath:
+ local_folder = os.path.dirname(file_path)
+ local_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, local_folder, local_file, self.remote_filepath)
+ elif isinstance(self.remote_filepath, list) and isinstance(self.local_filepath, str):
+ for file_path in self.remote_filepath:
+ remote_folder = os.path.dirname(file_path)
+ remote_file = os.path.basename(file_path)
+ file_msg = file_path
+ self._transfer(sftp_client, self.local_filepath, remote_file, remote_folder)
+ elif isinstance(self.remote_filepath, str) and isinstance(self.local_filepath, str):
+ local_folder = os.path.dirname(self.local_filepath)
+ file_msg = self.local_filepath
+ self._transfer(
+ sftp_client,
+ local_folder,
+ self.local_filepath,
+ self.remote_filepath,
+ only_file=True,
)
- file_msg = f"from {self.local_filepath} to {self.remote_filepath}"
- self.log.info("Starting to transfer file %s", file_msg)
- sftp_client.put(self.local_filepath, self.remote_filepath, confirm=self.confirm)
+ elif self.local_folder and self.remote_folder:
+ if self.operation.lower() == SFTPOperation.PUT:
+ files_list = self._search_files(os.listdir(self.local_folder))
+ for file in files_list:
+ local_file = os.path.basename(file)
+ file_msg = file
+ self._transfer(sftp_client, self.local_folder, local_file, self.remote_folder)
+ elif self.operation.lower() == SFTPOperation.GET:
+ files_list = self._search_files(sftp_client.listdir(self.remote_folder))
+ for file in files_list:
+ remote_file = ntpath.basename(file)
Review comment:
yep, sorry, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-873415902
Seems like very useful set of functionality. Could you please add unit tests for those @AngryHelper ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper closed pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper closed pull request #16792:
URL: https://github.com/apache/airflow/pull/16792
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-908647425
sorry for flood, I want create another PR with true commits
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-890806062
Need to fix linter issues. See guide: https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#contribution-workflow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AngryHelper closed pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
AngryHelper closed pull request #16792:
URL: https://github.com/apache/airflow/pull/16792
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #16792: [AIRFLOW-5703] SFTP Provider list of files and directories
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #16792:
URL: https://github.com/apache/airflow/pull/16792#issuecomment-873414449
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/BREEZE.rst) for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org