Fixes #24348: Strip URL scheme from hostPort at model construction

Centralise hostPort sanitisation in BaseModel.model_post_init() so that
every *Connection class automatically strips accidental http(s):// or
other URL-scheme prefixes the moment the model is constructed.

This replaces the previous approach of patching individual connector
connection.py files, covers all 38+ connectors at once, and removes the
need for per-connector call-site changes.

clean_host_port() is kept in db_utils.py as a public utility and is also
used inside get_host_from_host_port() as defence-in-depth.

Author: RajdeepKushwaha5

ingestion/src/metadata/ingestion/models/custom_pydantic.py

@@ -42,17 +42,24 @@ class BaseModel(PydanticBaseModel):
 
     def model_post_init(self, context: Any, /):
         """
-        This function is used to parse the FilterPattern fields for the Connection classes.
-        This is needed because dict is defined in the JSON schema for the FilterPattern field,
-        but a FilterPattern object is required in the generated code.
+        Post-init hook for Connection classes:
+        - Sanitises ``hostPort`` by stripping accidental URL scheme prefixes.
+        - Converts raw ``dict`` values into ``FilterPattern`` objects.
         """
         # pylint: disable=import-outside-toplevel
         try:
             if not self.__class__.__name__.endswith("Connection"):
-                # Only parse FilterPattern for Connection classes
                 return
             if not hasattr(self, "__pydantic_fields__"):
                 return
+
+            if "hostPort" in self.__pydantic_fields__:
+                raw = getattr(self, "hostPort", None)
+                if raw and "://" in raw:
+                    from metadata.utils.db_utils import clean_host_port
+
+                    object.__setattr__(self, "hostPort", clean_host_port(raw))
+
             for field in self.__pydantic_fields__:
                 if field.endswith("FilterPattern"):
                     from metadata.generated.schema.type.filterPattern import (

ingestion/src/metadata/utils/db_utils.py

@@ -12,9 +12,11 @@
 """
 Helpers module for db sources
 """
+
 import time
 import traceback
 from typing import Iterable, List, Union
+from urllib.parse import urlparse
 
 from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
 from metadata.generated.schema.entity.data.table import Table
@@ -43,12 +45,52 @@
 PUBLIC_SCHEMA = "public"
 
 
+def clean_host_port(host_port: str) -> str:
+    """
+    Strip URL scheme prefixes from a hostPort string.
+
+    Users sometimes enter a full URL (e.g. 'http://localhost:3306')
+    instead of just 'localhost:3306'. This strips the scheme to avoid
+    ValueError when parsing host and port.
+    """
+    host_port = host_port.strip()
+    if "://" not in host_port:
+        return host_port.rstrip("/")
+
+    logger.warning(
+        "The hostPort '%s' contains a URL scheme. "
+        "Expected format is 'hostname[:port]' (e.g. 'localhost:3306'). "
+        "Stripping the scheme prefix.",
+        host_port,
+    )
+    parsed = urlparse(host_port)
+    hostname = parsed.hostname or ""
+    try:
+        port = parsed.port
+    except ValueError as exc:
+        raise ValueError(
+            f"Invalid hostPort '{host_port}'. Expected format is "
+            "'hostname[:port]' (e.g. 'localhost:3306')."
+        ) from exc
+
+    if not hostname:
+        # urlparse couldn't extract hostname (e.g. jdbc:postgresql://host:5432)
+        # Fall back to stripping everything before the last ://
+        raw = host_port.rsplit("://", 1)[-1]
+        raw = raw.split("/", 1)[0]
+        raw = raw.split("?", 1)[0]
+        raw = raw.split("#", 1)[0]
+        return raw
+
+    return f"{hostname}:{port}" if port else hostname
+
+
 def get_host_from_host_port(uri: str) -> str:
     """
     if uri is like "localhost:9000"
     then return the host "localhost"
     """
-    return uri.split(":")[0]
+    return clean_host_port(uri).split(":")[0]
 
 
 #  pylint: disable=too-many-locals

ingestion/tests/unit/test_build_connection_url.py

@@ -97,6 +97,34 @@ def test_get_connection_url_mysql(self):
                 "mysql+pymysql://openmetadata_user:mocked_token@localhost:3306/openmetadata_db",
             )
 
+    def test_get_connection_url_mysql_with_url_scheme(self):
+        """hostPort with http:// prefix should be cleaned automatically"""
+        connection = MysqlConnectionConfig(
+            username="openmetadata_user",
+            authType=BasicAuth(password="openmetadata_password"),
+            hostPort="http://localhost:3306",
+            databaseSchema="openmetadata_db",
+        )
+        engine_connection = MySQLConnection(connection).client
+        self.assertEqual(
+            engine_connection.url.render_as_string(hide_password=False),
+            "mysql+pymysql://openmetadata_user:openmetadata_password@localhost:3306/openmetadata_db",
+        )
+
+    def test_get_connection_url_postgres_with_url_scheme(self):
+        """hostPort with https:// prefix should be cleaned automatically"""
+        connection = PostgresConnectionConfig(
+            username="openmetadata_user",
+            authType=BasicAuth(password="openmetadata_password"),
+            hostPort="https://localhost:5432",
+            database="openmetadata_db",
+        )
+        engine_connection = PostgresConnection(connection).client
+        self.assertEqual(
+            engine_connection.url.render_as_string(hide_password=False),
+            "postgresql+psycopg2://openmetadata_user:openmetadata_password@localhost:5432/openmetadata_db",
+        )
+
     def test_get_connection_url_postgres(self):
         connection = PostgresConnectionConfig(
             username="openmetadata_user",

ingestion/tests/unit/test_db_utils.py

@@ -12,6 +12,7 @@
 """
 Unit tests for db_utils module
 """
+
 import uuid
 from copy import deepcopy
 from unittest import TestCase
@@ -36,7 +37,11 @@
 from metadata.ingestion.lineage.models import Dialect
 from metadata.ingestion.lineage.sql_lineage import search_cache
 from metadata.ingestion.source.models import TableView
-from metadata.utils.db_utils import get_host_from_host_port, get_view_lineage
+from metadata.utils.db_utils import (
+    clean_host_port,
+    get_host_from_host_port,
+    get_view_lineage,
+)
 
 
 # Mock LineageTable class to simulate collate_sqllineage.core.models.Table
@@ -118,6 +123,63 @@ def test_get_host_from_host_port(self):
         self.assertEqual(get_host_from_host_port("localhost"), "localhost")
         self.assertEqual(get_host_from_host_port("example.com"), "example.com")
 
+        # Test with URL scheme prefixes
+        self.assertEqual(get_host_from_host_port("http://localhost:3306"), "localhost")
+        self.assertEqual(
+            get_host_from_host_port("https://example.com:5432"), "example.com"
+        )
+        self.assertEqual(get_host_from_host_port("http://localhost"), "localhost")
+
+    def test_clean_host_port(self):
+        """Test clean_host_port strips URL scheme prefixes"""
+        # Already-clean values pass through unchanged
+        self.assertEqual(clean_host_port("localhost:3306"), "localhost:3306")
+        self.assertEqual(clean_host_port("127.0.0.1:5432"), "127.0.0.1:5432")
+        self.assertEqual(clean_host_port("example.com"), "example.com")
+
+        # HTTP prefix is stripped
+        self.assertEqual(clean_host_port("http://localhost:3306"), "localhost:3306")
+        self.assertEqual(clean_host_port("http://example.com:8080"), "example.com:8080")
+
+        # HTTPS prefix is stripped
+        self.assertEqual(clean_host_port("https://localhost:5432"), "localhost:5432")
+        self.assertEqual(
+            clean_host_port("https://mydb.example.com:3306"), "mydb.example.com:3306"
+        )
+
+        # Trailing slash is stripped
+        self.assertEqual(clean_host_port("http://localhost:3306/"), "localhost:3306")
+
+        # Host only with scheme
+        self.assertEqual(clean_host_port("http://localhost"), "localhost")
+        self.assertEqual(clean_host_port("https://example.com"), "example.com")
+
+        # URL with path is handled — path/query/fragment are discarded
+        self.assertEqual(clean_host_port("http://localhost:3306/db"), "localhost:3306")
+        self.assertEqual(
+            clean_host_port("https://example.com:5432/mydb?ssl=true"),
+            "example.com:5432",
+        )
+
+        # Whitespace is stripped
+        self.assertEqual(clean_host_port("  localhost:3306  "), "localhost:3306")
+        self.assertEqual(clean_host_port("  http://localhost:3306  "), "localhost:3306")
+
+        # JDBC-style URLs fall back to raw extraction
+        self.assertEqual(clean_host_port("jdbc:postgresql://host:5432"), "host:5432")
+        self.assertEqual(clean_host_port("jdbc:postgresql://host:5432/db"), "host:5432")
+        self.assertEqual(
+            clean_host_port("jdbc:postgresql://host:5432?ssl=true"), "host:5432"
+        )
+        self.assertEqual(
+            clean_host_port("jdbc:postgresql://host:5432/db?ssl=true#ref"),
+            "host:5432",
+        )
+
+        # Invalid port raises ValueError
+        with self.assertRaises(ValueError):
+            clean_host_port("http://localhost:abc")
+
     @patch("metadata.utils.db_utils.ConnectionTypeDialectMapper")
     @patch("metadata.utils.db_utils.fqn")
     def test_get_view_lineage_success_with_lineage_parser(