Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/common/headers.py: 87%

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. 

17from __future__ import annotations 

18 

19from typing import Annotated 

20 

21from fastapi import Depends, Header, HTTPException, status 

22 

23from airflow.api_fastapi.common.types import Mimetype 

24 

25 

26def header_accept_json_or_text_depends( 

27 accept: Annotated[ 

28 str, 

29 Header( 

30 json_schema_extra={ 

31 "type": "string", 

32 "enum": [Mimetype.JSON, Mimetype.TEXT, Mimetype.ANY], 

33 } 

34 ), 

35 ] = Mimetype.ANY, 

36) -> Mimetype: 

37 if accept.startswith(Mimetype.ANY): 

38 return Mimetype.ANY 

39 if accept.startswith(Mimetype.JSON): 

40 return Mimetype.JSON 

41 if accept.startswith(Mimetype.TEXT): 

42 return Mimetype.TEXT 

43 raise HTTPException( 

44 status_code=status.HTTP_406_NOT_ACCEPTABLE, 

45 detail="Only application/json or text/plain is supported", 

46 ) 

47 

48 

49HeaderAcceptJsonOrText = Annotated[Mimetype, Depends(header_accept_json_or_text_depends)] 

50 

51 

52def header_accept_json_or_ndjson_depends( 

53 accept: Annotated[ 

54 str, 

55 Header( 

56 json_schema_extra={ 

57 "type": "string", 

58 "enum": [Mimetype.JSON, Mimetype.NDJSON, Mimetype.ANY], 

59 } 

60 ), 

61 ] = Mimetype.ANY, 

62) -> Mimetype: 

63 if accept.startswith(Mimetype.ANY): 

64 return Mimetype.ANY 

65 if accept.startswith(Mimetype.JSON): 

66 return Mimetype.JSON 

67 if accept.startswith(Mimetype.NDJSON) or accept.startswith(Mimetype.ANY): 

68 return Mimetype.NDJSON 

69 

70 raise HTTPException( 

71 status_code=status.HTTP_406_NOT_ACCEPTABLE, 

72 detail="Only application/json or application/x-ndjson is supported", 

73 ) 

74 

75 

76HeaderAcceptJsonOrNdjson = Annotated[Mimetype, Depends(header_accept_json_or_ndjson_depends)] 

77 

78 

79def header_content_type_json_or_form_depends( 

80 content_type: Annotated[ 

81 str, 

82 Header( 

83 alias="Content-Type", 

84 description="Content-Type of the request body", 

85 json_schema_extra={"enum": [Mimetype.JSON, Mimetype.FORM]}, 

86 ), 

87 ] = Mimetype.JSON, 

88) -> Mimetype: 

89 if content_type.startswith(Mimetype.JSON): 89 ↛ 91line 89 didn't jump to line 91 because the condition on line 89 was always true

90 return Mimetype.JSON 

91 if content_type.startswith(Mimetype.FORM): 

92 return Mimetype.FORM 

93 raise HTTPException( 

94 status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, 

95 detail="Only application/json or application/x-www-form-urlencoded is supported", 

96 ) 

97 

98 

99HeaderContentTypeJsonOrForm = Annotated[Mimetype, Depends(header_content_type_json_or_form_depends)]