refactor sync_bundles_to_db (apache#57150)

Author: rawwar

airflow-core/src/airflow/dag_processing/bundles/manager.py

@@ -243,17 +243,17 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
                 if not team:
                     raise _bundle_item_exc(f"Team '{config.team_name}' does not exist")
 
+            new_template, new_params = _extract_and_sign_template(name)
+
             if bundle := stored.pop(name, None):
                 bundle.active = True
-                new_template, new_params = _extract_and_sign_template(name)
                 if new_template != bundle.signed_url_template:
                     bundle.signed_url_template = new_template
                     self.log.debug("Updated URL template for bundle %s", name)
                 if new_params != bundle.template_params:
                     bundle.template_params = new_params
                     self.log.debug("Updated template parameters for bundle %s", name)
             else:
-                new_template, new_params = _extract_and_sign_template(name)
                 bundle = DagBundleModel(name=name)
                 bundle.signed_url_template = new_template
                 bundle.template_params = new_params
@@ -279,12 +279,13 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
                 )
                 bundle.teams = []
 
+        # Import here to avoid circular import
+        from airflow.models.errors import ParseImportError
+
         for name, bundle in stored.items():
             bundle.active = False
             bundle.teams = []
             self.log.warning("DAG bundle %s is no longer found in config and has been disabled", name)
-            from airflow.models.errors import ParseImportError
-
             session.execute(delete(ParseImportError).where(ParseImportError.bundle_name == name))
             self.log.info("Deleted import errors for bundle %s which is no longer configured", name)