Add example hook

democorp_airflow/hooks/example.py

@@ -0,0 +1,42 @@
+from airflow.hooks.base import BaseHook
+from airflow.models import Connection
+
+
+class ExampleHook(BaseHook):
+    """..."""
+
+    def __init__(self, conn_id: str) -> None:
+        """
+        Example Airflow hook. Implement your own business logic here.
+
+        :param conn_id: Airflow connection id
+        """
+        super().__init__()
+        self._conn_id = conn_id
+
+        self._conn: Connection | None = None
+
+    @property
+    def conn(self) -> Connection:
+        """
+        Cache connection to avoid re-fetching multiple times.
+
+        :return: Airflow connection object
+        """
+        if self._conn is None:
+            self._conn = self.get_connection(conn_id=self._conn_id)
+        return self._conn
+
+    def test_connection(self) -> tuple[bool, str]:
+        """
+        Verify a connection.
+
+        :return:
+        """
+        try:
+            # ... Implement your business logic to validate a connection here ...
+            _ = self.conn
+            return True, "Connection successful."
+        except Exception as e:
+            self.log.warning("Failed connection verification.")
+            return False, str(e)

tests/democorp_airflow/hooks/test_example.py

@@ -0,0 +1,24 @@
+import json
+from unittest import mock
+
+from democorp_airflow.hooks.example import ExampleHook
+
+
+@mock.patch.dict(
+    "os.environ",
+    AIRFLOW_CONN_MYDB=json.dumps(
+        {
+            "conn_type": "my-conn-type",
+            "login": "my-login",
+            "password": "my-password",
+            "host": "my-host",
+            "port": 1234,
+            "schema": "my-schema",
+            "extra": {"param1": "val1", "param2": "val2"},
+        }
+    ),
+)
+def test_examplehook():
+    """This test verifies fetching of a connection."""
+    test_hook = ExampleHook(conn_id="mydb")
+    test_hook.test_connection()