Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/public/plugins.py: 91%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17 

18from __future__ import annotations 

19 

20import structlog 

21from fastapi import Depends 

22from pydantic import ValidationError 

23 

24from airflow import plugins_manager 

25from airflow.api_fastapi.auth.managers.models.resource_details import AccessView 

26from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset 

27from airflow.api_fastapi.common.router import AirflowRouter 

28from airflow.api_fastapi.core_api.datamodels.plugins import ( 

29 PluginCollectionResponse, 

30 PluginImportErrorCollectionResponse, 

31 PluginResponse, 

32) 

33from airflow.api_fastapi.core_api.security import requires_access_view 

34 

35logger = structlog.get_logger(__name__) 

36 

37plugins_router = AirflowRouter(tags=["Plugin"], prefix="/plugins") 

38 

39 

40@plugins_router.get( 

41 "", 

42 dependencies=[Depends(requires_access_view(AccessView.PLUGINS))], 

43) 

44def get_plugins( 

45 limit: QueryLimit, 

46 offset: QueryOffset, 

47) -> PluginCollectionResponse: 

48 plugins_info = sorted(plugins_manager.get_plugin_info(), key=lambda x: x["name"]) 

49 valid_plugins: list[PluginResponse] = [] 

50 for plugin_dict in plugins_info: 

51 try: 

52 # Validate each plugin individually 

53 plugin = PluginResponse.model_validate(plugin_dict) 

54 valid_plugins.append(plugin) 

55 except ValidationError as e: 

56 logger.warning( 

57 "Skipping invalid plugin due to error", 

58 plugin_name=plugin_dict.get("name", "<unknown>"), 

59 error=str(e), 

60 ) 

61 continue 

62 

63 offset_value = offset.value or 0 

64 limit_value = limit.value if limit.value is not None else len(valid_plugins) 

65 

66 paginated_plugins = valid_plugins[offset_value : offset_value + limit_value] 

67 return PluginCollectionResponse( 

68 plugins=paginated_plugins, 

69 total_entries=len(valid_plugins), 

70 ) 

71 

72 

73@plugins_router.get( 

74 "/importErrors", 

75 dependencies=[Depends(requires_access_view(AccessView.PLUGINS))], 

76) 

77def import_errors() -> PluginImportErrorCollectionResponse: 

78 import_errors = plugins_manager.get_import_errors() 

79 return PluginImportErrorCollectionResponse.model_validate( 

80 { 

81 "import_errors": [{"source": source, "error": error} for source, error in import_errors.items()], 

82 "total_entries": len(import_errors), 

83 } 

84 )