Connection 관리하기

Connection 관리하기 (Managing Connections)

외부 서비스에 연결하는 데 필요한 자격 증명과 정보를 저장하는 Airflow의 Connection 객체를 관리하는 방법을 설명하는 문서예요. 환경 변수, Secrets Backend, 메타데이터 데이터베이스 세 가지 저장 방식과 JSON·URI 시리얼라이즈 형식, 커스텀 connection 타입 정의, connection 테스트까지 살펴볼게요.

출처: 문서

본문

더 보기 (See also)

hooks와 connections의 개요는 Connections & Hooks를 참고하세요.

Airflow의 Connection 객체는 외부 서비스에 연결하는 데 필요한 자격 증명과 기타 정보를 저장하는 데 사용돼요.

Connections는 다음 방식으로 정의할 수 있어요:

환경 변수에

외부 Secrets Backend에

Airflow 메타데이터 데이터베이스에 (CLI 또는 웹 UI 사용)

환경 변수에 connections 저장하기 (Storing connections in environment variables)

Airflow connections는 환경 변수에 정의할 수 있어요.

명명 규칙은 AIRFLOW_CONN_{CONN_ID}로, 전부 대문자예요(CONN을 단일 밑줄로 감싸는 점 참고). 따라서 connection id가 my_prod_db라면 변수 이름은 AIRFLOW_CONN_MY_PROD_DB가 돼요.

값은 JSON이거나 Airflow의 URI 형식일 수 있어요.

JSON 형식 예시 (JSON format example)

2.3.0 버전에 추가됨 (Added in version 2.3.0).

JSON으로 시리얼라이즈한다면:

export AIRFLOW_CONN_MY_PROD_DATABASE='{
    "conn_type": "my-conn-type",
    "login": "my-login",
    "password": "my-password",
    "host": "my-host",
    "port": 1234,
    "schema": "my-schema",
    "extra": {
        "param1": "val1",
        "param2": "val2"
    }
}'

JSON connection 표현 생성하기 (Generating a JSON connection representation)

2.8.0 버전에 추가됨 (Added in version 2.8.0).

connection JSON 생성을 더 쉽게 하기 위해 Connection 클래스에는 편리한 속성 as_json()이 있어요. 이렇게 사용할 수 있어요:

>>> from airflow.sdk import Connection
>>> c = Connection(
...     conn_id="some_conn",
...     conn_type="mysql",
...     description="connection description",
...     host="myhost.com",
...     login="myname",
...     password="mypassword",
...     extra={"this_param": "some val", "that_param": "other val*"},
... )
>>> print(f"AIRFLOW_CONN_{c.conn_id.upper()}='{c.as_json()}'")
AIRFLOW_CONN_SOME_CONN='{"conn_type": "mysql", "description": "connection description", "host": "myhost.com", "login": "myname", "password": "mypassword", "extra": {"this_param": "some val", "that_param": "other val*"}}'

또한 같은 접근 방식으로 Connection을 URI 형식에서 JSON 형식으로 변환할 수 있어요.

>>> from airflow.sdk import Connection
>>> c = Connection(
...     conn_id="awesome_conn",
...     description="Example Connection",
...     uri="aws://YOUR_AWS_ACCESS_KEY_ID:YOUR_AWS_SECRET_ACCESS_KEY@/?__extra__=%7B%22region_name%22%3A+%22eu-central-1%22%2C+%22config_kwargs%22%3A+%7B%22retries%22%3A+%7B%22mode%22%3A+%22standard%22%2C+%22max_attempts%22%3A+10%7D%7D%7D",
... )
>>> print(f"AIRFLOW_CONN_{c.conn_id.upper()}='{c.as_json()}'")
AIRFLOW_CONN_AWESOME_CONN='{"conn_type": "aws", "description": "Example Connection", "host": "", "login": "YOUR_AWS_ACCESS_KEY_ID", "password": "YOUR_AWS_SECRET_ACCESS_KEY", "schema": "", "extra": {"region_name": "eu-central-1", "config_kwargs": {"retries": {"mode": "standard", "max_attempts": 10}}}}'

URI 형식 예시 (URI format example)

Airflow URI로 시리얼라이즈한다면:

export AIRFLOW_CONN_MY_PROD_DATABASE='my-conn-type://login:password@host:port/schema?param1=val1&param2=val2'

유효한 URI를 생성하는 방법에 대한 자세한 내용은 아래 'Connection URI 형식'을 참고하세요.

UI와 CLI에서의 가시성 (Visibility in UI and CLI)

환경 변수로 정의된 connections는 Airflow UI에 표시되지 않고, airflow connections list로도 나열되지 않아요.

이는 이 connections가 런타임에 동적으로 해석되며, 일반적으로 태스크를 실행하는 worker 프로세스에서 해석되기 때문이에요. 이것들은 메타데이터 데이터베이스에 저장되지 않고 웹서버나 스케줄러 환경에도 로드되지 않아요.

이는 환경 기반 시크릿(예: .env 파일, Docker, Kubernetes secrets)을 users에게 노출되는 컴포넌트(웹서버)가 아니라 worker 같은 런타임 컴포넌트에만 주입하는 보안 배포 패턴을 지원해요.

가시성이나 편집을 위해 connections가 UI에 표시되길 원한다면 메타데이터 데이터베이스를 사용해 정의하세요.

Secrets Backend에 connections 저장하기 (Storing connections in a Secrets Backend)

HashiCorp Vault, AWS SSM Parameter Store 같은 외부 secrets backend에 Airflow connections를 저장할 수 있어요. 자세한 내용은 Secrets Backend를 참고하세요.

데이터베이스에 connections 저장하기 (Storing connections in the database)

더 보기 (See also)

Connections는 환경 변수나 HashiCorp Vault, AWS SSM Parameter Store 등의 외부 secrets backend에도 저장할 수 있어요.

connections를 데이터베이스에 저장할 때는 웹 UI나 Airflow CLI로 관리할 수 있어요.

UI로 Connection 만들기 (Creating a Connection with the UI)

UI의 Admin->Connections 섹션을 열어요. Add Connection 링크를 클릭해 새 connection을 만들어요.

Connection Id 필드에 원하는 connection ID를 채워요. 소문자를 사용하고 단어를 밑줄로 구분하는 것을 권장해요.
Connection Type 필드로 connection 타입을 고르세요.
나머지 필드를 채워요. 각 connection 타입에 속하는 필드에 대한 설명은 아래 'extra의 임의 dict 처리'를 참고하세요.
Save 버튼을 클릭해 connection을 만들어요.

UI로 Connection 편집하기 (Editing a Connection with the UI)

UI의 Admin->Connections 섹션을 열어요. connection 목록에서 편집하려는 connection 옆의 연필 아이콘을 클릭해요.

connection 속성을 수정하고 Save 버튼을 클릭해 변경 사항을 저장해요.

CLI로 Connection 만들기 (Creating a Connection from the CLI)

CLI에서 데이터베이스에 connection을 추가할 수 있어요.

JSON 형식으로 connection을 추가할 수 있어요 (2.3.0 버전부터):

airflow connections add 'my_prod_db' \
    --conn-json '{
        "conn_type": "my-conn-type",
        "login": "my-login",
        "password": "my-password",
        "host": "my-host",
        "port": 1234,
        "schema": "my-schema",
        "extra": {
            "param1": "val1",
            "param2": "val2"
        }
    }'

또는 Airflow Connection URI 형식을 사용할 수 있어요 (아래 'Connection URI 생성' 참고).

airflow connections add 'my_prod_db' \
    --conn-uri '<conn-type>://<login>:<password>@<host>:<port>/<schema>?param1=val1&param2=val2&...'

마지막으로 각 파라미터를 개별로 지정할 수도 있어요:

airflow connections add 'my_prod_db' \
    --conn-type 'my-conn-type' \
    --conn-login 'login' \
    --conn-password 'password' \
    --conn-host 'host' \
    --conn-port 'port' \
    --conn-schema 'schema' \
    ...

connections를 파일로 내보내기 (Exporting connections to file)

데이터베이스에 저장된 connections를 파일로 내보낼 수 있어요(예: 한 환경에서 다른 환경으로 connections를 마이그레이션할 때). 사용법은 Exporting Connections를 참고하세요.

데이터베이스 connections의 보안 (Security of connections in the database)

Airflow 메타데이터 데이터베이스에 저장된 connections에 대해 Airflow는 Fernet을 사용해 password와 기타 잠재적으로 민감한 데이터를 암호화해요. 이는 암호화 비밀번호가 없으면 key 없이 Connection Passwords를 조작하거나 읽을 수 없다는 것을 보장해요. Fernet 구성에 대한 정보는 Fernet을 참고하세요.

connections 테스트하기 (Testing Connections)

보안상의 이유로, test connection 기능은 Airflow UI·API·CLI 전반에서 기본적으로 비활성화돼 있어요.

사용자 기능에 대한 자세한 내용은 문서를 참고하세요: https://airflow.apache.org/docs/apache-airflow/stable/security/security_model.html#capabilities-of-authenticated-ui-users. "edit connection" 권한이 있는 신뢰도 높은 UI/API 사용자만 있는지 확인하기 전까지는 이 기능을 활성화하지 않는 것을 강력히 권장해요.

기능의 가용성은 Airflow 구성(airflow.cfg)의 core 섹션에 있는 test_connection 플래그로 제어할 수 있어요. 환경 변수 AIRFLOW__CORE__TEST_CONNECTION으로도 제어할 수 있어요.

이 구성 파라미터에 허용되는 값은 다음과 같아요:

Disabled: test connection 기능을 비활성화하고 UI의 Test Connection 버튼을 비활성화해요. Airflow 구성에 설정된 기본값이기도 해요.
Enabled: test connection 기능을 활성화하고 UI의 Test Connection 버튼을 활성화해요.
Hidden: test connection 기능을 비활성화하고 UI의 Test Connection 버튼을 숨겨요.

Test Connection을 활성화한 후에는 UI의 create/edit connection 페이지, Connections REST API 호출, 또는 airflow connections test CLI 명령 실행을 통해 사용할 수 있어요.

경고 (Warning)

Airflow UI나 REST API를 사용할 때 이 기능은 외부 secrets backend에 있는 connections에는 사용할 수 없어요.

connection을 테스트하려면 Airflow는 관련 hook 클래스의 test_connection 메서드를 호출하고 결과를 보고해요. connection 타입에 관련 hook이 없거나 hook에 test_connection 메서드 구현이 없을 수 있는데, 어느 경우든 오류 메시지가 표시되거나(UI에서 테스트 중이라면) 기능이 비활성화돼요.

참고 (Note)

Airflow UI에서 테스트할 때 테스트는 webserver에서 실행되므로 이 기능은 webserver에 설정된 네트워크 이그레스 규칙의 영향을 받아요.

참고 (Note)

webserver와 worker 머신(Airflow UI로 테스트할 때) 또는 머신/pod(Airflow CLI로 테스트할 때)에 다른 libs나 providers가 설치되어 있다면 테스트 결과가 다를 수 있어요.

비동기 (worker-디스패치) connection 테스트 (Asynchronous (worker-dispatched) connection testing)

위에서 설명한 테스트는 API 서버에서 실행돼요. Airflow는 대신 worker에서 테스트를 실행할 수도 있는데, 그러면 connection 자격 증명이 API 서버가 아니라 태스크가 사용하는 바로 그 곳인 worker에서만 사용돼요. 이는 connection이 worker에서만 도달 가능할 때, 또는 API 서버에서 자격 증명을 사용하고 싶지 않을 때 유용해요.

이것은 동기 테스트와 같은 [core] test_connection 활성화 플래그를 사용하고 Connections REST API를 통해 구동돼요: POST /connections/enqueue-test(토큰 반환)로 테스트를 제출한 다음, Airflow-Connection-Test-Token 헤더에 토큰을 전달하며 GET /connections/enqueue-test로 결과를 폴링해요. 결과는 그 토큰으로만, 그리고 connection에 권한이 있는 사용자만 읽을 수 있어요. 멀티팀 배포에서 다른 팀이 소유한 connection에 대한 테스트는 보이지 않아요.

테스트를 실행하는 worker는 위의 폴링 토큰과 별도로 인가돼요. 스케줄러는 단일 요청에 대해 수명이 짧은 JWT를 worker에 발급하는데, 이 JWT의 subject는 connection-test 요청 id이고 scope는 workload예요. worker가 호출하는 Execution API 엔드포인트는 ct:self 체크를 강제해요 — 토큰 subject가 요청 경로의 connection-test id와 일치해야 하므로, worker의 토큰은 발급된 하나의 요청에 대해서만 connection을 가져오고 결과를 보고할 수 있어요. 다른 connection 테스트, task instance, connection에는 도달할 수 없어요.

이 동작은 [connection_test] 섹션에서 조정돼요:

connection_test.timeout — 테스트가 타임아웃으로 보고되기 전에 실행될 수 있는 시간(기본 60초).
connection_test.max_concurrency — 동시에 실행될 수 있는 테스트 수. 추가 요청은 슬롯이 생길 때까지 대기해요(기본 4).
connection_test.reaper_interval — Airflow가 타임아웃을 초과한 테스트를 확인하고 실패 처리하는 주기(기본 30.0초).

참고 (Note)

테스트가 worker에서 실행되므로 그 결과는 해당 worker가 사용할 수 있는 라이브러리·providers·네트워크 접근을 반영하며, API 서버와 다를 수 있어요.

커스텀 connection 타입 (Custom connection types)

Airflow는 커스텀 connection 타입의 정의를 허용해요 — connections의 add/edit 폼 수정을 포함해요. 커스텀 connection 타입은 커뮤니티 유지보수 providers에 정의되어 있고, 커스텀 connection 타입을 추가하는 커스텀 provider를 추가할 수도 있어요. 커스텀 providers를 추가하는 방법에 대한 설명은 Providers를 참고하세요.

provider.yaml의 connection-types 배열로 connection 타입을 노출함으로써 Airflow를 다음과 같이 커스터마이징할 수 있어요:

커스텀 connection 타입 추가
connection 타입에서 자동 Hook 생성 추가
connection URL에서 커스텀 "extra" 파라미터를 표시·편집하는 커스텀 폼 필드 추가
connection에 사용되지 않는 표준 필드 숨기기
필드가 어떻게 포맷되어야 하는지 보여주는 placeholders 추가

커스텀 providers를 추가하는 방법에 대한 자세한 내용은 Providers에서 더 읽을 수 있어요.

커스텀 connection 필드 (Custom connection fields)

참고 (Note)

선호 접근 방식: provider.yaml에 connection UI 메타데이터 정의

Airflow 3.2부터 커스텀 connection 필드와 필드 동작을 정의하는 선호 방식은 provider.yaml에서 선언적으로 하는 것이에요. 이 접근 방식은 런타임에 flask_appbuilder나 wtforms를 import할 필요가 없어요.

Python hook 메서드 get_connection_form_widgets()와 get_ui_field_behaviour()는 폴백으로 계속 동작하며 deprecation 공지와 마이그레이션 기간 후에만 제거될 거예요. 옛 접근 방식으로 작성된 커스텀 providers는 계속 동작할 거예요. 다만 새 providers는 아래 설명하는 YAML 접근 방식을 사용해야 해요.

provider.yaml에서 connection UI 메타데이터 정의하기 (Defining connection UI metadata in provider.yaml)

Connection 폼 메타데이터는 provider의 provider.yaml 파일에서 connection-types 아래에 선언적으로 정의돼요. 두 섹션이 있어요:

conn-fields — Connection.extra에 저장되는 커스텀 필드:

connection-types:
  - hook-class-name: airflow.providers.myservice.hooks.myservice.MyServiceHook
    connection-type: myservice
    conn-fields:
      workspace:
        label: Workspace
        schema:
          type:
            - string
            - 'null'
      project:
        label: Project ID
        schema:
          type:
            - string
            - 'null'

ui-field-behaviour — 표준 connection 필드에 대한 커스터마이징(숨기기, 레이블 변경, placeholders):

connection-types:
  - hook-class-name: airflow.providers.myservice.hooks.myservice.MyServiceHook
    connection-type: myservice
    ui-field-behaviour:
      hidden-fields:
        - port
        - host
        - login
        - schema
      relabeling:
        password: API Token
      placeholders:
        password: your-api-token
        workspace: My workspace gid
        project: My project gid

필드 스키마 타입은 JSON Schema 규칙을 따르고 있어요.

지원되는 conn-fields 스키마 옵션의 전체 참조는 Use Params to Provide a Trigger UI Form를 참고하세요.

Python으로 connection UI 메타데이터 정의하기 (legacy) (Defining connection UI metadata in Python (legacy))

참고 (Note)

아래 Python 메서드 접근 방식은 계속 동작하고 deprecation 공지 없이는 제거되지 않아요. 다만 새 providers는 위에서 설명한 YAML 접근 방식을 사용해야 해요.

Hook 클래스에 get_connection_form_widgets()를 구현해 connection add/edit 뷰에 커스텀 폼 필드를 추가할 수 있어요. 키는 extra dict에 저장되어야 하는 필드의 문자열 이름이어야 해요. 값은 wtforms.fields.core.Field의 상속체여야 해요.

예시는 다음과 같아요:

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

    return {
        "workspace": StringField(lazy_gettext("Workspace"), widget=BS3TextFieldWidget()),
        "project": StringField(lazy_gettext("Project"), widget=BS3TextFieldWidget()),
    }

참고 (Note)

커스텀 필드는 더 이상 extra__<conn type>__ 프리픽스가 필요하지 않아요

Airflow 2.3 이전에는 UI에서 커스텀 필드를 원한다면 extra__<conn type>__로 프리픽스해야 했고, 이렇게 extra dict에 값이 저장됐어요. 2.3부터는 더 이상 그럴 필요가 없어요.

get_ui_field_behaviour() 메서드는 표준 필드의 동작을 커스터마이징할 수 있게 해요. 예를 들어 필드를 숨기거나 레이블을 다시 달거나(사용되지 않거나 용도가 바뀐 경우) placeholder 텍스트를 추가할 수 있어요.

예시는 다음과 같아요:

@staticmethod
def get_ui_field_behaviour() -> dict[str, Any]:
    """Returns custom field behaviour"""
    return {
        "hidden_fields": ["port", "host", "login", "schema"],
        "relabeling": {},
        "placeholders": {
            "password": "Asana personal access token",
            "workspace": "My workspace gid",
            "project": "My project gid",
        },
    }

참고 (Note)

표준 connection 속성(즉 login, password, host, scheme, port, extra)과 이름이 충돌하는 extra 필드에 폼 placeholder를 추가하려면 extra__<conn type>__로 프리픽스해야 해요. 예: extra__myservice__password.

무엇을 할 수 있는지에 대한 예시는 providers를 참고하세요. 예: JdbcHook.

참고 (Note)

사용 중단된 hook-class-names

Airflow 2.2.0 이전에는 providers의 connections가 provider 메타데이터의 hook-class-names 배열로 노출됐어요. 하지만 이는 worker에서 개별 hooks를 사용할 때 비효율적임이 밝혀졌고, hook-class-names 배열은 이제 connection-types 배열로 대체됐어요. provider가 2.2.0 미만의 Airflow를 지원할 때까지는 connection-types와 hook-class-names가 모두 존재해야 해요. CI 빌드 중 자동 검사가 이 두 배열의 일관성을 검증해요.

URI 형식 (URI format)

참고 (Note)

2.3.0 버전부터는 connections를 JSON으로 시리얼라이즈할 수 있어요. 위 예시를 참고하세요.

역사적인 이유로 Airflow에는 Connection 객체를 문자열 값으로 시리얼라이즈하는 데 사용할 수 있는 특별한 URI 형식이 있어요.

일반적으로 Airflow의 URI 형식은 다음과 같아요:

my-conn-type://my-login:my-password@my-host:5432/my-schema?param1=val1&param2=val2

위 URI는 다음에 해당하는 Connection 객체를 만들어요:

Connection(
    conn_id="",
    conn_type="my_conn_type",
    description=None,
    login="my-login",
    password="my-password",
    host="my-host",
    port=5432,
    schema="my-schema",
    extra=json.dumps(dict(param1="val1", param2="val2")),
)

Connection URI 생성하기 (Generating a connection URI)

connection URI 생성을 더 쉽게 하기 위해 Connection 클래스에는 편리한 메서드 get_uri()가 있어요. 이렇게 사용할 수 있어요:

>>> import json
>>> from airflow.sdk import Connection
>>> c = Connection(
...     conn_id="some_conn",
...     conn_type="mysql",
...     description="connection description",
...     host="myhost.com",
...     login="myname",
...     password="mypassword",
...     extra=json.dumps(dict(this_param="some val", that_param="other val*")),
... )
>>> print(f"AIRFLOW_CONN_{c.conn_id.upper()}='{c.get_uri()}'")
AIRFLOW_CONN_SOME_CONN='mysql://myname:***@myhost.com?this_param=some+val&that_param=other+val%2A'

참고 (Note)

get_uri() 메서드는 SQLAlchemy 호환 URI가 아닌 Airflow 형식의 connection URI를 반환해요. 데이터베이스 connections에 SQLAlchemy 호환 URI가 필요하면 sqlalchemy_url 속성을 대신 사용하세요.

또한 connection을 만들었다면 airflow connections get 명령을 사용할 수 있어요.

$ airflow connections get sqlite_default
Id: 40
Connection Id: sqlite_default
Connection Type: sqlite
Host: /tmp/sqlite_default.db
Schema: null
Login: null
Password: null
Port: null
Is Encrypted: false
Is Extra Encrypted: false
Extra: {}
URI: sqlite://%2Ftmp%2Fsqlite_default.db

extra의 임의 dict 처리 (Handling of arbitrary dict in extra)

일부 JSON 구조는 손실 없이 urlencode할 수 없어요. 그런 JSON에 대해 get_uri는 전체 문자열을 url 쿼리 파라미터 __extra__ 아래에 저장해요.

예를 들어:

>>> extra_dict = {"my_val": ["list", "of", "values"], "extra": {"nested": {"json": "val"}}}
>>> c = Connection(
...     conn_type="scheme",
...     host="host/location",
...     schema="schema",
...     login="user",
...     password="password",
...     port=1234,
...     extra=json.dumps(extra_dict),
... )
>>> uri = c.get_uri()
>>> uri
'scheme://user:password@host%2Flocation:1234/schema?__extra__=%7B%22my_val%22%3A+%5B%22list%22%2C+%22of%22%2C+%22values%22%5D%2C+%22extra%22%3A+%7B%22nested%22%3A+%7B%22json%22%3A+%22val%22%7D%7D%7D'

그리고 같은 딕셔너리를 반환하는지 확인할 수 있어요:

>>> new_c = Connection(uri=uri)
>>> new_c.extra_dejson == extra_dict
True

하지만 가장 흔한 key-value 쌍만 저장하는 경우에는 일반 url encoding이 사용돼요.

URI가 올바르게 파싱되는지 이렇게 확인할 수 있어요:

>>> from airflow.sdk import Connection

>>> c = Connection(uri="my-conn-type://my-login:my-password@my-host:5432/my-schema?param1=val1&param2=val2")
>>> print(c.login)
my-login
>>> print(c.password)
my-password

connection 파라미터의 특수 문자 처리 (Handling of special characters in connection params)

참고 (Note)

connection을 생성할 때는 'Connection URI 생성' 섹션에서 설명한 편리한 메서드 Connection.get_uri를 사용하세요. 이 섹션은 정보 제공 목적만을 위한 것이에요.

URI를 수동으로 만들 때는 특정 문자에 특별한 처리가 필요해요.

예를 들어 password에 /가 있으면 실패해요:

>>> c = Connection(uri="my-conn-type://my-login:my-pa/ssword@my-host:5432/my-schema?param1=val1&param2=val2")
ValueError: invalid literal for int() with base 10: 'my-pa'

이를 고치려면 quote_plus()로 인코딩할 수 있어요:

>>> c = Connection(uri="my-conn-type://my-login:my-pa%2Fssword@my-host:5432/my-schema?param1=val1&param2=val2")
>>> print(c.password)
my-pa/ssword

Connection 관리하기