Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/hooks/filesystem.py: 46%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
20from pathlib import Path
21from typing import Any
23from airflow.hooks.base import BaseHook
26class FSHook(BaseHook):
27 """
28 Allows for interaction with an file server.
30 Connection should have a name and a path specified under extra:
32 example:
33 Connection Id: fs_test
34 Connection Type: File (path)
35 Host, Schema, Login, Password, Port: empty
36 Extra: {"path": "/tmp"}
37 """
39 conn_name_attr = "fs_conn_id"
40 default_conn_name = "fs_default"
41 conn_type = "fs"
42 hook_name = "File (path)"
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
51 return {"path": StringField(lazy_gettext("Path"), widget=BS3TextFieldWidget())}
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 }
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
68 def get_conn(self) -> None:
69 pass
71 def get_path(self) -> str:
72 """
73 Get the path to the filesystem location.
75 :return: the path.
76 """
77 return self.basepath
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)