fix(ingestion): keep custom_pydantic dependency-free to avoid circular import

The model_post_init hook added in the previous commit imported
`clean_host_port` from `metadata.utils.db_utils` at runtime. That
module has a heavy dependency graph (OpenMetadata client, LineageParser,
generated schemas), so every *Connection class construction would drag
it in and cause circular-import failures during integration test
bootstrap.

The file docstring explicitly requires BaseModel to be 'self-sufficient
with only pydantic at import time', so the hostPort sanitisation is now
handled by a stdlib-only helper (_strip_hostport_scheme) colocated with
BaseModel. `clean_host_port` remains the public API in db_utils and
delegates to the same helper, preserving behaviour for all existing
callers and unit tests.

Author: RajdeepKushwaha5

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

@@ -18,6 +18,7 @@
 import json
 import logging
 from typing import Any, Callable, Dict, Literal, Optional, Union
+from urllib.parse import urlparse
 
 from pydantic import BaseModel as PydanticBaseModel
 from pydantic import WrapSerializer, model_validator
@@ -34,6 +35,55 @@
 JSON_ENCODERS = "json_encoders"
 
 
+def _strip_hostport_scheme(raw: str) -> str:
+    """
+    Strip an accidental URL scheme from a hostPort string.
+
+    Self-contained helper that depends only on the standard library so
+    ``model_post_init`` never drags heavy ``metadata.*`` imports into the
+    bootstrap path of every generated Connection class.
+
+    Raises ValueError if the scheme carries a non-numeric port so the user
+    gets a clear error instead of a silently broken hostPort.
+    """
+    value = raw.strip()
+    if "://" not in value:
+        return value
+
+    parsed = urlparse(value)
+    hostname = parsed.hostname or ""
+    safe_label = (
+        f"{parsed.scheme}://{hostname}"
+        if parsed.scheme and hostname
+        else "URL with scheme"
+    )
+    logger.warning(
+        "The hostPort '%s' contains a URL scheme. Expected format is "
+        "'hostname[:port]' (e.g. 'localhost:3306'). Stripping the scheme prefix.",
+        safe_label,
+    )
+    try:
+        port = parsed.port
+    except ValueError as exc:
+        raise ValueError(
+            f"Invalid hostPort '{safe_label}'. Expected format is "
+            "'hostname[:port]' (e.g. 'localhost:3306')."
+        ) from exc
+
+    if not hostname:
+        # urlparse couldn't extract a hostname (e.g. 'jdbc:postgresql://host:5432/db')
+        # Fall back to stripping scheme and any trailing path/query/fragment/userinfo.
+        tail = value.rsplit("://", 1)[-1]
+        for sep in ("/", "?", "#"):
+            tail = tail.split(sep, 1)[0]
+        if "@" in tail:
+            tail = tail.rsplit("@", 1)[-1]
+        return tail
+
+    host = f"[{hostname}]" if ":" in hostname else hostname
+    return f"{host}:{port}" if port is not None else host
+
+
 class BaseModel(PydanticBaseModel):
     """
     Base model for OpenMetadata generated models.
@@ -55,12 +105,10 @@ def model_post_init(self, context: Any, /):
         if "hostPort" in self.__pydantic_fields__:
             raw = getattr(self, "hostPort", None)
             if isinstance(raw, str) and "://" in raw:
-                from metadata.utils.db_utils import clean_host_port
-
-                # Let ValueError propagate: if clean_host_port cannot parse
-                # the input (e.g. non-numeric port), the user must fix their
-                # config rather than silently getting a broken hostPort.
-                object.__setattr__(self, "hostPort", clean_host_port(raw))
+                # Let ValueError propagate: if the hostPort cannot be parsed
+                # (e.g. non-numeric port), the user must fix their config
+                # rather than silently getting a broken hostPort.
+                object.__setattr__(self, "hostPort", _strip_hostport_scheme(raw))
 
         try:
             for field in self.__pydantic_fields__:

ingestion/src/metadata/utils/db_utils.py

@@ -16,7 +16,6 @@
 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
@@ -34,6 +33,7 @@
     get_lineage_by_query,
     get_lineage_via_table_entity,
 )
+from metadata.ingestion.models.custom_pydantic import _strip_hostport_scheme
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.models import TableView
 from metadata.utils import fqn
@@ -52,45 +52,14 @@ def clean_host_port(host_port: str) -> str:
     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.
+
+    Delegates to the stdlib-only helper colocated with ``BaseModel`` so the
+    behaviour stays in lockstep with Pydantic's ``model_post_init`` hook.
     """
-    host_port = host_port.strip()
-    if "://" not in host_port:
-        return host_port.rstrip("/")
-
-    parsed = urlparse(host_port)
-    hostname = parsed.hostname or ""
-    safe_label = (
-        f"{parsed.scheme}://{hostname}"
-        if parsed.scheme and hostname
-        else "URL with scheme"
-    )
-    logger.warning(
-        "The hostPort '%s' contains a URL scheme. "
-        "Expected format is 'hostname[:port]' (e.g. 'localhost:3306'). "
-        "Stripping the scheme prefix.",
-        safe_label,
-    )
-    try:
-        port = parsed.port
-    except ValueError as exc:
-        raise ValueError(
-            f"Invalid hostPort '{safe_label}'. 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]
-        if "@" in raw:
-            raw = raw.rsplit("@", 1)[-1]
-        return raw
-
-    host = f"[{hostname}]" if ":" in hostname else hostname
-    return f"{host}:{port}" if port is not None else host
+    value = host_port.strip()
+    if "://" not in value:
+        return value.rstrip("/")
+    return _strip_hostport_scheme(value)
 
 
 def get_host_from_host_port(uri: str) -> str: