Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/ui/dependencies.py: 35%

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 

20from typing import Literal 

21 

22from fastapi import Depends, status 

23from fastapi.exceptions import HTTPException 

24 

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

26from airflow.api_fastapi.common.db.common import SessionDep 

27from airflow.api_fastapi.common.router import AirflowRouter 

28from airflow.api_fastapi.core_api.datamodels.ui.common import BaseGraphResponse 

29from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc 

30from airflow.api_fastapi.core_api.security import ReadableDagsFilterDep, requires_access_dag 

31from airflow.api_fastapi.core_api.services.ui.dependencies import ( 

32 extract_single_connected_component, 

33 get_data_dependencies, 

34 get_scheduling_dependencies, 

35) 

36 

37dependencies_router = AirflowRouter(tags=["Dependencies"]) 

38 

39 

40@dependencies_router.get( 

41 "/dependencies", 

42 responses=create_openapi_http_exception_doc( 

43 [ 

44 status.HTTP_400_BAD_REQUEST, 

45 status.HTTP_404_NOT_FOUND, 

46 ] 

47 ), 

48 dependencies=[Depends(requires_access_dag("GET", DagAccessEntity.DEPENDENCIES))], 

49) 

50def get_dependencies( 

51 session: SessionDep, 

52 readable_dags_filter: ReadableDagsFilterDep, 

53 node_id: str | None = None, 

54 dependency_type: Literal["scheduling", "data"] = "scheduling", 

55) -> BaseGraphResponse: 

56 """Dependencies graph.""" 

57 if dependency_type == "data": 

58 if node_id is None or not node_id.startswith("asset:"): 

59 raise HTTPException( 

60 status.HTTP_400_BAD_REQUEST, "Data dependencies require an asset node_id (e.g., 'asset:123')" 

61 ) 

62 

63 try: 

64 asset_id = int(node_id.replace("asset:", "")) 

65 except ValueError: 

66 raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset node_id: {node_id}") 

67 

68 data = get_data_dependencies(asset_id, session, readable_dags_filter.value) 

69 if not data["nodes"]: 

70 raise HTTPException(status.HTTP_404_NOT_FOUND, f"Asset with id {asset_id} was not found") 

71 return BaseGraphResponse(**data) 

72 

73 data = get_scheduling_dependencies(readable_dags_filter.value) 

74 

75 if node_id is not None: 

76 try: 

77 data = extract_single_connected_component(node_id, data["nodes"], data["edges"]) 

78 except ValueError as e: 

79 raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) 

80 

81 return BaseGraphResponse(**data)