airflow yandex 源码

airflow yandex 代码

文件路径：/airflow/providers/yandex/hooks/yandex.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import json
import warnings
from typing import Any

import yandexcloud

from airflow.exceptions import AirflowException
from airflow.hooks.base import BaseHook


class YandexCloudBaseHook(BaseHook):
    """
    A base hook for Yandex.Cloud related tasks.

    :param yandex_conn_id: The connection ID to use when fetching connection info.
    """

    conn_name_attr = 'yandex_conn_id'
    default_conn_name = 'yandexcloud_default'
    conn_type = 'yandexcloud'
    hook_name = 'Yandex Cloud'

    @staticmethod
    def get_connection_form_widgets() -> dict[str, Any]:
        """Returns connection widgets to add to connection form"""
        from flask_appbuilder.fieldwidgets import BS3PasswordFieldWidget, BS3TextFieldWidget
        from flask_babel import lazy_gettext
        from wtforms import PasswordField, StringField

        return {
            "extra__yandexcloud__service_account_json": PasswordField(
                lazy_gettext('Service account auth JSON'),
                widget=BS3PasswordFieldWidget(),
                description='Service account auth JSON. Looks like '
                '{"id", "...", "service_account_id": "...", "private_key": "..."}. '
                'Will be used instead of OAuth token and SA JSON file path field if specified.',
            ),
            "extra__yandexcloud__service_account_json_path": StringField(
                lazy_gettext('Service account auth JSON file path'),
                widget=BS3TextFieldWidget(),
                description='Service account auth JSON file path. File content looks like '
                '{"id", "...", "service_account_id": "...", "private_key": "..."}. '
                'Will be used instead of OAuth token if specified.',
            ),
            "extra__yandexcloud__oauth": PasswordField(
                lazy_gettext('OAuth Token'),
                widget=BS3PasswordFieldWidget(),
                description='User account OAuth token. '
                'Either this or service account JSON must be specified.',
            ),
            "extra__yandexcloud__folder_id": StringField(
                lazy_gettext('Default folder ID'),
                widget=BS3TextFieldWidget(),
                description='Optional. This folder will be used '
                'to create all new clusters and nodes by default',
            ),
            "extra__yandexcloud__public_ssh_key": StringField(
                lazy_gettext('Public SSH key'),
                widget=BS3TextFieldWidget(),
                description='Optional. This key will be placed to all created Compute nodes'
                'to let you have a root shell there',
            ),
        }

    @classmethod
    def provider_user_agent(cls) -> str | None:
        """Construct User-Agent from Airflow core & provider package versions"""
        import airflow
        from airflow.providers_manager import ProvidersManager

        try:
            manager = ProvidersManager()
            provider_name = manager.hooks[cls.conn_type].package_name  # type: ignore[union-attr]
            provider = manager.providers[provider_name]
            return f'apache-airflow/{airflow.__version__} {provider_name}/{provider.version}'
        except KeyError:
            warnings.warn(f"Hook '{cls.hook_name}' info is not initialized in airflow.ProviderManager")
            return None

    @staticmethod
    def get_ui_field_behaviour() -> dict[str, Any]:
        """Returns custom field behaviour"""
        return {
            "hidden_fields": ['host', 'schema', 'login', 'password', 'port', 'extra'],
            "relabeling": {},
        }

    def __init__(
        self,
        # Connection id is deprecated. Use yandex_conn_id instead
        connection_id: str | None = None,
        yandex_conn_id: str | None = None,
        default_folder_id: str | None = None,
        default_public_ssh_key: str | None = None,
    ) -> None:
        super().__init__()
        if connection_id:
            warnings.warn(
                "Using `connection_id` is deprecated. Please use `yandex_conn_id` parameter.",
                DeprecationWarning,
                stacklevel=2,
            )
        self.connection_id = yandex_conn_id or connection_id or self.default_conn_name
        self.connection = self.get_connection(self.connection_id)
        self.extras = self.connection.extra_dejson
        credentials = self._get_credentials()
        self.sdk = yandexcloud.SDK(user_agent=self.provider_user_agent(), **credentials)
        self.default_folder_id = default_folder_id or self._get_field('folder_id', False)
        self.default_public_ssh_key = default_public_ssh_key or self._get_field('public_ssh_key', False)
        self.client = self.sdk.client

    def _get_credentials(self) -> dict[str, Any]:
        service_account_json_path = self._get_field('service_account_json_path', False)
        service_account_json = self._get_field('service_account_json', False)
        oauth_token = self._get_field('oauth', False)
        if not (service_account_json or oauth_token or service_account_json_path):
            raise AirflowException(
                'No credentials are found in connection. Specify either service account '
                'authentication JSON or user OAuth token in Yandex.Cloud connection'
            )
        if service_account_json_path:
            with open(service_account_json_path) as infile:
                service_account_json = infile.read()
        if service_account_json:
            service_account_key = json.loads(service_account_json)
            return {'service_account_key': service_account_key}
        else:
            return {'token': oauth_token}

    def _get_field(self, field_name: str, default: Any = None) -> Any:
        """Fetches a field from extras, and returns it."""
        long_f = f'extra__yandexcloud__{field_name}'
        if hasattr(self, 'extras') and long_f in self.extras:
            return self.extras[long_f]
        else:
            return default

相关信息

airflow 源码目录

相关文章

airflow init 源码

airflow yandexcloud_dataproc 源码