Airflow 뜯어보기(2)

# airflow\\configuration.py
def getboolean(self, section: str, key: str, **kwargs) -> bool:  # type: ignore[override]
    val = str(self.get(section, key, _extra_stacklevel=1, **kwargs)).lower().strip()
    if "#" in val:
        val = val.split("#")[0].strip()
    if val in ("t", "true", "1"):
        return True
    elif val in ("f", "false", "0"):
        return False
    else:
        raise AirflowConfigException(
            f'Failed to convert value to bool. Please check "{key}" key in "{section}" section. '
            f'Current value: "{val}".'
        )

저번 글에서 getboolean 함수를 살펴보다가, get 함수에 대해 알아보기로 했다.

# airflow\\configuration.py
@overload  # type: ignore[override]
def get(self, section: str, key: str, fallback: str = ..., **kwargs) -> str: ...

@overload  # type: ignore[override]
def get(self, section: str, key: str, **kwargs) -> str | None: ...

def get(  # type: ignore[override,misc]
    self,
    section: str,
    key: str,
    suppress_warnings: bool = False,
    _extra_stacklevel: int = 0,
    **kwargs,
) -> str | None:
    section = section.lower()
    key = key.lower()
    warning_emitted = False
    deprecated_section: str | None
    deprecated_key: str | None

    option_description = self.configuration_description.get(section, {}).get(key, {})
    ...

먼저, 위 코드에서 get 함수가 여러 번 선언된 이유가 궁금했다.

@overload  # type: ignore[override]
def get(self, section: str, key: str, fallback: str = ..., **kwargs) -> str: ...

@overload  # type: ignore[override]
def get(self, section: str, key: str, **kwargs) -> str | None: ...

def get(  # type: ignore[override,misc]
    self,
    section: str,
    key: str,
    suppress_warnings: bool = False,
    _extra_stacklevel: int = 0,
    **kwargs,
) -> str | None:
    ...

파이썬에서 위와 같은 코드 구조는 타입 힌트와 함수 오버로딩을 명확히 하기 위해 사용된다. 실제로 함수가 여러 번 선언되는 것이 아니라, @overload 데코레이터를 사용해 타입 힌트를 정의하고, 마지막에 실제 구현체를 제공하는 방식이다.

@overload는 타입 힌트를 위한 함수 서명을 정의할 때 사용된다. 실제로는 실행되지 않으며, 정적 타입 검사기에서만 사용된다. 즉, @overload로 정의된 함수는 런타임에 존재하지 않으며, 타입 체크를 도와주는 역할만 하는 것이다.

따라서 fallback이 없을 때는 str 또는 None 타입이 반환되고, fallback이 있을 때는 str 타입이 반환됨을 명시한다. 마지막으로 fallback이 없고 suppress_warnings와 _extra_stacklevel이 있으면 str 또는 None을 반환한다.

그 다음 단계로 option_description = self.configuration_description.get(section, {}).get(key, {})를 살펴보자.

configuration_description은 retrieve_configuration_description 함수로 정의된다.

def retrieve_configuration_description(
    include_airflow: bool = True,
    include_providers: bool = True,
    selected_provider: str | None = None,
) -> dict[str, dict[str, Any]]:
    """
    Read Airflow configuration description from YAML file.

    :param include_airflow: Include Airflow configs
    :param include_providers: Include provider configs
    :param selected_provider: If specified, include selected provider only
    :return: Python dictionary containing configs & their info
    """
    base_configuration_description: dict[str, dict[str, Any]] = {}
    if include_airflow:
        with open(_default_config_file_path("config.yml")) as config_file:
            base_configuration_description.update(yaml.safe_load(config_file))
    ...

주석을 보면 Airflow 설정, provide 설정 등에 관련된 파라미터가 존재하고 이에 따라 딕셔너리를 반환하는 것을 알 수 있다.

저번 글을 보면 core와 unit_test_mode를 각각 section과 key로 매핑하여 넣어줬으므로, configuration_description 에서 core에 해당하는 부분을 가져온 뒤, unit_test_mode에 해당하는 부분마저 가져온다는 것을 유추할 수 있다.

이 부분에 해당하는 내용을 config.yml 파일에서 찾아보면 내용은 다음과 같다.

unit_test_mode:
  description: |
    Turn unit test mode on (overwrites many configuration options with test
    values at runtime)
  version_added: ~
  type: string
  example: ~
  default: "False"

결론

__main__.py에서 첫 번째로 실행되는 코드가 conf = configuration.conf와 같을 때, configuration.py에서 첫 번째로 실행되는 conf: AirflowConfigParser = initialize_config()는 Airflow의 설정 파일을 읽고, unit_test_mode인지 확인한 뒤, 웹 서버 관련 설정을 하고 AirflowConfigParser 객체를 반환한다는 것을 알 수 있었다.

다음 글에서는 secrets_backend_list = initialize_secrets_backends()를 살펴보자..