Move default_celery.py to inside the provider (#32628)

This has been missed in #32526 and is extracted out from #32604
in an attempt to make it smaller and separately reviewable.

This one adds also deprecation warning to handle the
configuration value that people might already have in
the [celery] iccelery_config_options"

airflow/config_templates/__init__.py

@@ -15,3 +15,14 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
+from airflow.utils.deprecation_tools import add_deprecated_classes
+
+__deprecated_classes = {
+    "default_celery": {
+        "DEFAULT_CELERY_CONFIG": "airflow.providers.celery.executors.default_celery.DEFAULT_CELERY_CONFIG",
+    },
+}
+
+add_deprecated_classes(__deprecated_classes, __name__)

airflow/config_templates/config.yml

@@ -2145,7 +2145,7 @@ celery:
       version_added: ~
       type: string
       example: ~
-      default: "airflow.config_templates.default_celery.DEFAULT_CELERY_CONFIG"
+      default: "airflow.providers.celery.executors.default_celery.DEFAULT_CELERY_CONFIG"
     ssl_active:
       description: ~
       version_added: ~

airflow/config_templates/default_airflow.cfg

@@ -1109,7 +1109,7 @@ flower_basic_auth =
 sync_parallelism = 0
 
 # Import path for celery configuration options
-celery_config_options = airflow.config_templates.default_celery.DEFAULT_CELERY_CONFIG
+celery_config_options = airflow.providers.celery.executors.default_celery.DEFAULT_CELERY_CONFIG
 ssl_active = False
 
 # Path to the client key.

airflow/providers/celery/executors/celery_executor_utils.py

@@ -39,11 +39,11 @@
 from setproctitle import setproctitle
 
 import airflow.settings as settings
-from airflow.config_templates.default_celery import DEFAULT_CELERY_CONFIG
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
 from airflow.executors.base_executor import BaseExecutor
 from airflow.models.taskinstance import TaskInstanceKey
+from airflow.providers.celery.executors.default_celery import DEFAULT_CELERY_CONFIG
 from airflow.stats import Stats
 from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -64,6 +64,7 @@
 
 if conf.has_option("celery", "celery_config_options"):
     celery_configuration = conf.getimport("celery", "celery_config_options")
+
 else:
     celery_configuration = DEFAULT_CELERY_CONFIG
 

airflow/config_templates/default_celery.py → airflow/providers/celery/executors/default_celery.py

@@ -74,11 +74,16 @@ def _broker_supports_visibility_timeout(url):
     "worker_enable_remote_control": conf.getboolean("celery", "worker_enable_remote_control"),
 }
 
-celery_ssl_active = False
-try:
-    celery_ssl_active = conf.getboolean("celery", "SSL_ACTIVE")
-except AirflowConfigException:
-    log.warning("Celery Executor will run without SSL")
+
+def _get_celery_ssl_active() -> bool:
+    try:
+        return conf.getboolean("celery", "SSL_ACTIVE")
+    except AirflowConfigException:
+        log.warning("Celery Executor will run without SSL")
+        return False
+
+
+celery_ssl_active = _get_celery_ssl_active()
 
 try:
     if celery_ssl_active:

tests/providers/celery/executors/test_celery_executor.py

@@ -258,8 +258,7 @@ def test_cleanup_stuck_queued_tasks(self, mock_fail):
     def test_result_backend_sqlalchemy_engine_options(self, mock_celery):
         import importlib
 
-        from airflow.config_templates import default_celery
-        from airflow.providers.celery.executors import celery_executor_utils
+        from airflow.providers.celery.executors import celery_executor_utils, default_celery
 
         # reload celery conf to apply the new config
         importlib.reload(default_celery)