Airflow 뜯어보기(1)

먼저 Airflow의 시작점인 __main__.py 파일의 main() 함수부터 살펴보자.

# airflow\__main__.py
def main():
    conf = configuration.conf
    if conf.get("core", "security") == "kerberos":
        os.environ["KRB5CCNAME"] = conf.get("kerberos", "ccache")
        os.environ["KRB5_KTNAME"] = conf.get("kerberos", "keytab")
    parser = cli_parser.get_parser()
    argcomplete.autocomplete(parser)
    args = parser.parse_args()
    if args.subcommand not in ["lazy_loaded", "version"]:
        from airflow.configuration import write_default_airflow_configuration_if_needed

        conf = write_default_airflow_configuration_if_needed()
        if args.subcommand in ["webserver", "internal-api", "worker"]:
            write_webserver_configuration_if_needed(conf)
    configure_internal_api(args, conf)

    args.func(args)

 

main 함수의 첫 줄에 있는 conf 변수에는 어떤 값이 저장될까?

configuration.py를 살펴보니 마지막 3줄이 다음과 같았다.

# airflow\configuration.py
conf: AirflowConfigParser = initialize_config()
secrets_backend_list = initialize_secrets_backends()
conf.validate()

initialize_config 함수를 통해 Airflow 초기 설정 값을 파싱하고, secrets_backend_list 변수를 생성한 다음, validate 함수로 최종 검증을 하는 것 같다.

 

그렇다면 __main__.py에서 어떻게 configuration.py 파일에서 선언한 conf 변수를 사용할 수 있는 것일까?

__main__.py 파일의 상단 부분을 살펴보면 다음과 같이 configuration.py가 import 되어 있다.

from airflow import configuration

이는 모듈 레벨에서 실행되는 코드로, configuration.py가 처음 import될 때 자동으로 실행된다.

모듈 레벨에서 실행되는 코드가 무슨 말일까?

파이썬 파일이 모듈로 사용될 때, 최상위 수준에서 실행되는 코드를 의미한다.

파이썬은 파일을 스크립트로 실행하거나 모듈로 가져올 수 있는데, 이 두 가지 경우 모두 최상위 수준에서 작성된 코드이다.

 

나는 다음과 같이 테스트를 해봤다.

# b.py
def hello_function():
    return "Hello, World!"

hello = hello_function()

# a.py
import b

def main():
    hello = b.hello
    print(hello)

if __name__ == "__main__":
    main()

위와 같은 코드를 작성하고 a.py를 실행해보니, "Hello, World!"가 출력되는 것을 확인할 수 있었다.

 

그리고 다음과 같이 example.py 파일을 정의하고, a.py 파일에서 모듈로서 import하면 "This always runs (module-level code)"가 출력되고, main() 함수에 있는 코드는 출력되지 않음을 확인했다.

# example.py
print("This always runs (module-level code)")

def main():
    print("This only runs when executed directly")

if __name__ == "__main__":
    main()

# a.py
import b
import example

def main():
    p = b.conf
    print(p)

if __name__ == "__main__":
    main()

 

다시 돌아와서 configuration 파일의 conf 변수가 무엇을 가리키는 지 확인해보자.

# configuration.py
conf: AirflowConfigParser = initialize_config()

타입 힌트를 보면 알 수 있듯이 conf 변수에는 AirflowConfigParser객체가 저장되는 거 같다.

AirflowConfigParser 객체는 initialize_config() 함수를 통해 생성된다.

 

def initialize_config() -> AirflowConfigParser:
    """
    Load the Airflow config files.

    Called for you automatically as part of the Airflow boot process.
    """
    airflow_config_parser = AirflowConfigParser()
    if airflow_config_parser.getboolean("core", "unit_test_mode"):
        airflow_config_parser.load_test_config()
    else:
        load_standard_airflow_configuration(airflow_config_parser)
        # If the user set unit_test_mode in the airflow.cfg, we still
        # want to respect that and then load the default unit test configuration
        # file on top of it.
        if airflow_config_parser.getboolean("core", "unit_test_mode"):
            airflow_config_parser.load_test_config()
    # Set the WEBSERVER_CONFIG variable
    global WEBSERVER_CONFIG
    WEBSERVER_CONFIG = airflow_config_parser.get("webserver", "config_file")
    return airflow_config_parser

일단 airflow_config_parser 변수에 AirflowConfigParser() 객체를 담고, 그 후에는 getboolean 메서드를 사용하여 무언가를 하는 거 같다.

 

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의 파라미터로 들어온 "core"와 "unit_test_mode"를 다시 한 번 get 메서드로 넘겨준 다음에 그 리턴값을 val에 저장하는데…

get 메서드가 복잡해 보이니까, 다음 글에서 마저 살펴보도록 하자.