Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/airflow/hooks/filesystem.py: 46%

1# 

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

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

4# distributed with this work for additional information 

5# regarding copyright ownership. The ASF licenses this file 

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

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

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

9# 

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

11# 

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

13# software distributed under the License is distributed on an 

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

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

16# specific language governing permissions and limitations 

17# under the License. 

18from __future__ import annotations 

19 

20from pathlib import Path 

21from typing import Any 

22 

23from airflow.hooks.base import BaseHook 

24 

25 

26class FSHook(BaseHook): 

27 """ 

28 Allows for interaction with an file server. 

29 

30 Connection should have a name and a path specified under extra: 

31 

32 example: 

33 Connection Id: fs_test 

34 Connection Type: File (path) 

35 Host, Schema, Login, Password, Port: empty 

36 Extra: {"path": "/tmp"} 

37 """ 

38 

39 conn_name_attr = "fs_conn_id" 

40 default_conn_name = "fs_default" 

41 conn_type = "fs" 

42 hook_name = "File (path)" 

43 

44 @classmethod 

45 def get_connection_form_widgets(cls) -> dict[str, Any]: 

46 """Return connection widgets to add to connection form.""" 

47 from flask_appbuilder.fieldwidgets import BS3TextFieldWidget 

48 from flask_babel import lazy_gettext 

49 from wtforms import StringField 

50 

51 return {"path": StringField(lazy_gettext("Path"), widget=BS3TextFieldWidget())} 

52 

53 @classmethod 

54 def get_ui_field_behaviour(cls) -> dict[str, Any]: 

55 """Return custom field behaviour.""" 

56 return { 

57 "hidden_fields": ["host", "schema", "port", "login", "password", "extra"], 

58 "relabeling": {}, 

59 "placeholders": {}, 

60 } 

61 

62 def __init__(self, fs_conn_id: str = default_conn_name, **kwargs): 

63 super().__init__(**kwargs) 

64 conn = self.get_connection(fs_conn_id) 

65 self.basepath = conn.extra_dejson.get("path", "") 

66 self.conn = conn 

67 

68 def get_conn(self) -> None: 

69 pass 

70 

71 def get_path(self) -> str: 

72 """ 

73 Get the path to the filesystem location. 

74 

75 :return: the path. 

76 """ 

77 return self.basepath 

78 

79 def test_connection(self): 

80 """Test File connection.""" 

81 try: 

82 p = self.get_path() 

83 if not p: 

84 return False, "File Path is undefined." 

85 if not Path(p).exists(): 

86 return False, f"Path {p} does not exist." 

87 return True, f"Path {p} is existing." 

88 except Exception as e: 

89 return False, str(e)