Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/execution_api/routes/connections.py: 65%

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 logging 

21from typing import Annotated 

22 

23from fastapi import APIRouter, Depends, HTTPException, Path, status 

24 

25from airflow.api_fastapi.execution_api.datamodels.connection import ConnectionResponse 

26from airflow.api_fastapi.execution_api.security import CurrentTIToken, get_team_name_dep 

27from airflow.exceptions import AirflowNotFoundException 

28from airflow.models.connection import Connection 

29 

30 

31async def has_connection_access( 

32 connection_id: Annotated[str, Path(min_length=1)], 

33 token=CurrentTIToken, 

34) -> bool: 

35 """Check if the task has access to the connection.""" 

36 log.debug( 

37 "Checking access for task instance with key '%s' to connection '%s'", 

38 token.id, 

39 connection_id, 

40 ) 

41 

42 # The current version of Airflow does not support true 

43 # multi-tenancy yet (this is well-documented at 

44 # https://airflow.apache.org/docs/apache-airflow/stable/security/security_model.html#future-multi-tenancy-isolation), 

45 # so for now we always return 'True' here. 

46 # When we introduce true multi-tenancy in the future 

47 # this would be the place to do add a check. 

48 return True 

49 

50 

51router = APIRouter( 

52 responses={status.HTTP_404_NOT_FOUND: {"description": "Connection not found"}}, 

53 dependencies=[Depends(has_connection_access)], 

54) 

55 

56log = logging.getLogger(__name__) 

57 

58 

59@router.get( 

60 "/{connection_id}", 

61 responses={ 

62 status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"}, 

63 status.HTTP_403_FORBIDDEN: {"description": "Task does not have access to the connection"}, 

64 }, 

65) 

66def get_connection( 

67 connection_id: Annotated[str, Path(min_length=1)], 

68 team_name: Annotated[str | None, Depends(get_team_name_dep)], 

69) -> ConnectionResponse: 

70 """Get an Airflow connection.""" 

71 try: 

72 connection = Connection.get_connection_from_secrets(connection_id, team_name=team_name) 

73 except AirflowNotFoundException: 

74 raise HTTPException( 

75 status.HTTP_404_NOT_FOUND, 

76 detail={ 

77 "reason": "not_found", 

78 "message": f"Connection with ID {connection_id} not found", 

79 }, 

80 ) 

81 return ConnectionResponse.model_validate(connection)