`FTPSHook.store_file()` change directory

[makrushin-evgenii]

Apache Airflow version

2.7.2

What happened

FTPSHook.store_file() change current directory. And second call with same directory will raise no such file or directory error:

[2023-10-18, 14:40:08 MSK] {logging_mixin.py:149} INFO - content hash uploading to `test/daily/20230601_transactions.csv.MD5` ...
[2023-10-18, 14:40:08 MSK] {logging_mixin.py:149} INFO - content uploading to `test/daily/20230601_transactions.csv` ...
[2023-10-18, 14:40:08 MSK] {taskinstance.py:1824} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/decorators/base.py", line 220, in execute
    return_value = super().execute(context)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/operators/python.py", line 181, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/operators/python.py", line 198, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/repo/dags/integrations_alpharm_reporting_dag.py", line 59, in upload_external_shops_report
    upload_report_to_ftp(task_id, f'test/daily/{logical_date:YYYYMMDD}_transactions.csv')
  File "/opt/airflow/dags/repo/common/integrations_alpharm/utils.py", line 36, in upload_report_to_ftp
    from_drive2_to_ftp(get_report_drive2_path(task_id), ftp_path)
  File "/opt/airflow/dags/repo/common/integrations_alpharm/utils.py", line 32, in from_drive2_to_ftp
    ftp_hook.store_file(ftp_path, BytesIO(content))
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/ftp/hooks/ftp.py", line 220, in store_file
    conn.cwd(remote_path)
  File "/usr/local/lib/python3.10/ftplib.py", line 625, in cwd
    return self.voidcmd(cmd)
  File "/usr/local/lib/python3.10/ftplib.py", line 286, in voidcmd
    return self.voidresp()
  File "/usr/local/lib/python3.10/ftplib.py", line 259, in voidresp
    resp = self.getresp()
  File "/usr/local/lib/python3.10/ftplib.py", line 254, in getresp
    raise error_perm(resp)
ftplib.error_perm: 550 test/daily: Нет такого файла или каталога

This happens because of this line in store_file() implementation:

conn.cwd(remote_path)
conn.storbinary(f'STOR {remote_file_name}', input_handle)

To get around this, you have to recreate the FTPSHook for each uploading. It would be more convenient to simply restore directory in the FTPSHook.store_file() method after storbinary call

What you think should happen instead

No response

How to reproduce

ftp_hook = FTPSHook()
ftp_hook.get_conn().prot_p()  # https://stackoverflow.com/questions/65473257/ftpshook-airflow-522-ssl-tls-required-on-the-data-channel
ftp_hook.store_file(ftp_path, BytesIO(bytes)) # OK
ftp_hook.store_file(ftp_path, BytesIO(bytes)) # Raise "ftplib.error_perm: 550 test/daily: no such file or directory"

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[makrushin-evgenii]

I am willing to submit a PR this weekend, maybe next week

[makrushin-evgenii]

Submit a PR: #35105