프로그래매틱 호출

프로그래매틱 호출 (Programmatic invocations)

셸에서 실행하는 대신 Python 스크립트와 애플리케이션에서 dbt 명령을 호출할 수 있는 기능이에요. dbt CLI와 같은 명령 표면을 유지하면서 더 큰 애플리케이션이나 워크플로에 dbt 실행을 임베드할 때 유용해요.

출처: 문서

본문

프로그래매틱 호출을 사용하면 셸에서 실행하는 대신 Python 스크립트와 애플리케이션에서 dbt 명령을 호출할 수 있어요. 더 큰 애플리케이션이나 워크플로에 dbt 실행을 임베드하면서도 dbt CLI와 같은 명령 표면을 사용하고 싶을 때 유용해요.

일반적인 사용 사례:

• Python 애플리케이션이나 서비스의 일부로 dbt 실행
• 오케스트레이션 워크플로에 dbt 실행 통합
• dbt 명령을 실행하고 결과를 검사해야 하는 내부 도구 구축

아직 설치하지 않았다면 PyPI의 dbt v1 패키지를 참고해 dbt v1용 공식 Python 패키지를 설치하세요.

(dbt v2.0 이상에 적용)

from dbt.cli.main import dbtRunner, dbtRunnerResult

# initialize
dbt = dbtRunner()

# create CLI args as a list of strings
cli_args = ["run", "--select", "tag:my_tag"]

# run the command
res: dbtRunnerResult = dbt.invoke(cli_args)

# inspect the results
for r in res.result:
    print(f"{r.unique_id}: {r.status}")

구현 세부사항은 dbt v1 저장소의 dbt-python 크레이트를 참고하세요.

지원되는 인자 (Supported arguments)

dbtRunner.invoke는 dbt CLI와 같은 인자를 받아요. 첫 번째 위치 인자는 명령(예: run, build, test)이고, 그 뒤에 명령줄에서 보통 전달하는 플래그와 옵션이 따라와요.

예를 들어 dbt.invoke(["run", "--select", "tag:my_tag"])는 dbt run --select tag:my_tag를 실행하는 것과 같아요. 별도의 dbtRunner 특정 인자 목록은 없어요. 사용 가능한 옵션의 권위 있는 출처는 CLI 도움말(dbt --help, dbt run --help 등)과 dbt 명령 레퍼런스 문서예요.

from dbt.cli.main import dbtRunner

dbt = dbtRunner()

# equivalent ways to pass arguments
dbt.invoke(["run", "--select", "tag:my_tag"])
dbt.invoke(["run"], select="tag:my_tag")

병렬 실행 미지원 (Parallel execution not supported)

dbt는 같은 프로세스에서 여러 호출을 병렬로 안전하게 실행하는 것을 지원하지 않아요. 한 프로세스 안에서 여러 dbt 명령을 동시에 실행하는 것은 안전하지 않고 공식적으로 권장하지 않아요. 서브프로세스를 관리하는 래퍼 프로세스가 필요해요. 그 이유는:

• 동시에 실행되는 명령이 데이터 플랫폼과 예기치 않게 상호작용할 수 있어요. 예를 들어 같은 모델에 대해 dbt run과 dbt build를 동시에 실행하면 예측 불가능한 결과가 나올 수 있어요.
• 각 dbt 명령은 전역 Python 변수와 상호작용해요. 안전하게 동작하려면 명령을 별도 프로세스에서 실행해야 해요. 예를 들어 서브프로세스를 생성하거나 Celery로 오케스트레이션하세요.

안전한 병렬 실행을 위해 dbt 플랫폼 CLI나 Studio IDE를 사용할 수 있어요. 둘 다 사용자 대신 동시성(여러 프로세스)을 관리하는 추가 작업을 해줘요.

(dbt v2.0 이상) v2에서는 호출이 스레드 레벨 락으로 직렬화되어, 같은 프로세스 안에서 여러 호출이 동시에 실행될 수 없어요. (v1에서는 병렬 실행이 지원되지 않았지만 락이 없어서 호출이 멀티스레드 모드에서 여전히 실행될 수 있었어요.) v1과 마찬가지로 multiprocessing으로 각 호출을 별도 프로세스에서 실행해 병렬화할 수 있어요.

dbtRunnerResult

각 명령은 다음 속성을 가진 dbtRunnerResult 객체를 반환해요:

• success (bool): 명령이 성공했는지 여부.
• result: 명령이 완료되면(성공 또는 처리된 에러로) 명령의 결과를 반환해요. 반환 타입은 명령마다 달라요.
• exception: dbt 호출이 처리되지 않은 에러를 만나 완료되지 않았을 때 발생한 예외.
• catalog (v2 전용): 카탈로그 생성을 요청할 때 명령이 생성하는 카탈로그.

(dbt v2.0 이상) v2 엔진은 Rust로 구현돼서 exception은 더 이상 dbt가 발생시킨 정확한 Python 예외 객체를 담지 않아요. 대신 잡힌 에러 메시지가 예외 타입으로 전달돼요:

• 엔진이 호출을 받기 전에 외부 함수 인터페이스(FFI) 경계에서 호출이 실패하면 exception은 ValueError나 RuntimeError 같은 풀린(unwrapped) 예외 타입을 담아요.
• 엔진 내부에서 호출이 실패하면 exception은 DbtRunnerError예요.

v2는 또한 카탈로그 생성을 요청할 때 dbtRunnerResult에 최상위 catalog 속성을 추가해요. v1에서는 catalog.json이 dbt docs generate를 실행할 때만 생성됐어요. v2에서는 --write-catalog 플래그를 전달해 어떤 명령의 일부로든 카탈로그를 생성할 수 있어요. 예를 들어 dbt run --write-catalog은 dbtRunnerResult.result와 dbtRunnerResult.catalog를 모두 채워요.

CLI 종료 코드와 프로그래매틱 호출이 반환하는 dbtRunnerResult 사이에는 일대일 대응이 있어요:

약속과 주의사항 (Commitments and caveats)

우리는 dbt v1의 CLI와 기능적으로 동등한 Python 진입점을 제공하겠다는 지속적인 약속을 하고 있어요. 우리는 그 목표를 달성하는 데 사용하는 기본 구현을 바꿀 권리를 보유해요. 현재 구현이 단기·중기적으로 실제 사례를 열어줄 것으로 기대하며, 그 사이에 궁극적으로 대체할 안정적·장기적 인터페이스 세트를 작업 중이에요.

특히 각 명령이 dbtRunnerResult.result에서 반환하는 객체는 완전히 계약되지 않아 변경될 수 있어요. 반환되는 일부 객체는 dbt 아티팩트 내용과 일부 겹치기 때문에 부분적으로 문서화돼 있어요. Python 객체로서 직렬화된 JSON 아티팩트에서 사용할 수 있는 것보다 훨씬 많은 필드와 메서드를 담아요. 이러한 추가 필드와 메서드는 내부용으로 간주되어 dbt의 향후 버전에서 변경될 수 있어요.

고급 사용 패턴 (Advanced usage patterns)

⚠️ 주의: 이러한 패턴의 문법과 지원은 dbt의 향후 버전에서 변경될 수 있어요. dbtRunner의 목표는 프로그래매틱 환경에서 CLI 워크플로와의 동등성을 제공하는 거예요. CLI로는 불가능한 것을 확장하는 몇 가지 고급 사용 패턴이 있어요.

(dbt v2.0 이상) 객체 재사용 (Reusing objects) — v2에서는 매니페스트 주입(Manifest injection)이 지원되지 않아요. 미리 구성된 Manifest를 dbtRunner에 전달할 수 없어요.

콜백 등록 (Registering callbacks) — v2에서는 dbt의 EventManager에 콜백을 등록하는 것이 지원되지 않아요.

매개변수 재정의 (Overriding parameters) — CLI 스타일 문자열 목록 대신 키워드 인자로 매개변수를 전달하세요. 현재 dbt는 입력에 대한 검증이나 타입 강제 변환을 수행하지 않아요. 명령은 목록 안의 첫 번째 위치 인자로 지정해야 해요.

from dbt.cli.main import dbtRunner

dbt = dbtRunner()

# these are equivalent
dbt.invoke(["--fail-fast", "run", "--select", "tag:my_tag"])
dbt.invoke(["run"], select=["tag:my_tag"], fail_fast=True)

더 알아보기 (Learn more)

• dbt v1용 공식 Python 패키지는 PyPI의 dbt v1 패키지를 참고하세요.
• 명령 레퍼런스와 사용 가능한 플래그는 dbt Commands 문서를 참고하세요.