# Copyright 2025 Collate # Licensed under the Collate Community License, Version 1.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from airflow.plugins_manager import AirflowPlugin from flask import Blueprint from openmetadata_managed_apis.api.app import get_blueprint from openmetadata_managed_apis.api.config import PLUGIN_NAME from openmetadata_managed_apis.views.rest_api import RestApiView # Creating View to be used by Plugin rest_api_view = {"category": "Admin", "name": "REST API Plugin", "view": RestApiView()} # Creating template Blueprint template_blueprint = Blueprint( "template_blueprint", __name__, template_folder="views/templates", ) # Import REST API blueprint # In Airflow 2.x, we need cached_app context # In Airflow 3.x, cached_app doesn't exist, so we import directly try: from airflow.www.app import cached_app with cached_app().app_context(): api_blueprint = get_blueprint() except (ImportError, ModuleNotFoundError): # Airflow 3.x - no cached_app needed api_blueprint = get_blueprint() class RestApiPlugin(AirflowPlugin): """ Creating the RestApiPlugin which extends the AirflowPlugin to import it into Airflow Uses Flask Blueprints for both Airflow 2.x and 3.x. In Airflow 3.x, Flask blueprints are served through WSGI middleware at /pluginsv2/ prefix. """ name = PLUGIN_NAME operators = [] # noqa: RUF012 hooks = [] # noqa: RUF012 executors = [] # noqa: RUF012 menu_links = [] # noqa: RUF012 # Use Flask Blueprints for both Airflow 2.x and 3.x flask_blueprints = [template_blueprint, api_blueprint] if api_blueprint else [] appbuilder_views = [rest_api_view] if rest_api_view else [] admin_views = [rest_api_view] if rest_api_view else []