Allow FTPHook to change port number (#39465)

* Allow FTPHook to change port

* Add integration tests and fix a bug

* Styling fix

* Add logging of the FTP host and port

* Update logging and test styling

* styling fix

* formatting change

* Update airflow/providers/ftp/hooks/ftp.py

* fix log statement formatting

---------

Co-authored-by: Elad Kalif <45845474+eladkal@users.noreply.github.com>

airflow/providers/ftp/hooks/ftp.py

@@ -19,10 +19,13 @@
 
 import datetime
 import ftplib  # nosec: B402
+import logging
 from typing import Any, Callable
 
 from airflow.hooks.base import BaseHook
 
+logger = logging.getLogger(__name__)
+
 
 class FTPHook(BaseHook):
     """
@@ -58,7 +61,15 @@ def get_conn(self) -> ftplib.FTP:
         if self.conn is None:
             params = self.get_connection(self.ftp_conn_id)
             pasv = params.extra_dejson.get("passive", True)
-            self.conn = ftplib.FTP(params.host, params.login, params.password)  # nosec: B321
+            self.conn = ftplib.FTP()  # nosec: B321
+            if params.host:
+                port = ftplib.FTP_PORT
+                if params.port is not None:
+                    port = params.port
+                logger.info("Connecting via FTP to %s:%d", params.host, port)
+                self.conn.connect(params.host, port)
+                if params.login:
+                    self.conn.login(params.login, params.password)
             self.conn.set_pasv(pasv)
 
         return self.conn

tests/providers/ftp/hooks/test_ftp.py

@@ -155,6 +155,28 @@ def setup_method(self):
             Connection(conn_id="ftp_active", conn_type="ftp", host="localhost", extra='{"passive": false}')
         )
 
+        db.merge_conn(
+            Connection(
+                conn_id="ftp_custom_port",
+                conn_type="ftp",
+                host="localhost",
+                port=10000,
+                extra='{"passive": true}',
+            )
+        )
+
+        db.merge_conn(
+            Connection(
+                conn_id="ftp_custom_port_and_login",
+                conn_type="ftp",
+                host="localhost",
+                port=10000,
+                login="user",
+                password="pass123",
+                extra='{"passive": true}',
+            )
+        )
+
     def _test_mode(self, hook_type, connection_id, expected_mode):
         hook = hook_type(connection_id)
         conn = hook.get_conn()
@@ -172,6 +194,26 @@ def test_ftp_active_mode(self, mock_ftp):
 
         self._test_mode(FTPHook, "ftp_active", False)
 
+    @mock.patch("ftplib.FTP")
+    def test_ftp_custom_port(self, mock_ftp):
+        from airflow.providers.ftp.hooks.ftp import FTPHook
+
+        hook = FTPHook("ftp_custom_port")
+        conn = hook.get_conn()
+        conn.connect.assert_called_once_with("localhost", 10000)
+        conn.login.assert_not_called()
+        conn.set_pasv.assert_called_once_with(True)
+
+    @mock.patch("ftplib.FTP")
+    def test_ftp_custom_port_and_login(self, mock_ftp):
+        from airflow.providers.ftp.hooks.ftp import FTPHook
+
+        hook = FTPHook("ftp_custom_port_and_login")
+        conn = hook.get_conn()
+        conn.connect.assert_called_once_with("localhost", 10000)
+        conn.login.assert_called_once_with("user", "pass123")
+        conn.set_pasv.assert_called_once_with(True)
+
     @mock.patch("ftplib.FTP_TLS")
     def test_ftps_passive_mode(self, mock_ftp):
         from airflow.providers.ftp.hooks.ftp import FTPSHook