Python에서 소스와 인증 정보 사용하기

Python 변환, 외부 함수와 Compute Module은 초기화된 소스 객체를 통해 외부 시스템의 연결, 비밀 정보와 세션 인증 정보에 접근할 수 있습니다. 변환에서는 가져온 소스를 @external_systems에 선언하여 받으며, Python 함수는 구성한 소스 또는 별칭을 get_source()로 가져옵니다. Compute Module은 모듈에 연결한 소스를 해당 실행 환경의 소스 인터페이스로 제공합니다.

아래 함수들은 실행 환경이 제공한 실제 Source 객체를 인수로 받습니다. 소스 초기화나 인증에 성공한 것처럼 가정하는 빈 객체를 만들지 않습니다. 변환에서 소스를 가져오는 전체 절차는 외부 연결 변환에 있습니다.

HTTPS 연결과 기본 클라이언트

REST 기반 소스는 Python requests의 Session을 바탕으로 한 클라이언트를 제공합니다. 소스에 설정한 서버 및 클라이언트 인증서와 프록시 연결 구성이 반영됩니다.

HTTPS 연결은 정확히 하나여야 합니다. 연결이 없거나 둘 이상이면 Only single connection sources are supported. 오류가 발생하며 기본 클라이언트를 만들 수 없습니다. 실행 환경의 External connection 측면 패널에서 연결 수를 확인합니다.

다음 함수는 지정한 상대 경로를 조회하고 실제 응답 문자열을 반환합니다. 상대 경로는 /로 시작해야 합니다. HTTP 오류는 호출자에게 전달합니다.

from external_systems.sources import HttpsConnection, Source
from requests import Session


def get_source_text(source: Source, relative_path: str) -> str:
    connection: HttpsConnection = source.get_https_connection()
    client: Session = connection.get_client()
    response = client.get(connection.url + relative_path, timeout=10)
    response.raise_for_status()
    return response.text

예를 들어 Dictionary API 소스에는 /api/v2/entries/en/hello를 전달할 수 있습니다. 소스의 기본 URL과 상대 경로를 구분합니다. HTTPS 클라이언트를 쓰기 전이나 사용 중 os.chdir()로 작업 디렉터리를 바꾸면 보안 연결에 필요한 환경 변수의 상대 경로 참조가 깨질 수 있습니다.

비밀 정보

소스의 비밀 정보 키는 설정 화면이나 실행 환경의 소스 패널에서 확인합니다. get_secret()은 해당 키의 실제 문자열 값을 반환합니다. 키가 없거나 유효한 값을 제공하지 못하면 오류가 발생합니다.

from external_systems.sources import Source


def read_source_secret(source: Source, secret_name: str) -> str:
    return source.get_secret(secret_name)

반환한 비밀 정보를 로그나 출력 데이터셋에 기록하지 않습니다. 필요한 클라이언트 인증 구성에 직접 전달합니다.

세션 인증 정보와 갱신

정적 비밀 정보 대신 실행 시 생성하는 인증 정보는 get_session_credentials()로 받습니다. 반환값은 Refreshable[SourceCredentials]이며 get()이 현재 인증 정보를 제공합니다.

지원 여부는 소스와 실행 환경에 따릅니다. 세션 인증 정보가 없는 소스에 이 메서드를 호출하면 ValueError가 발생합니다. 갱신 가능한 제공자도 실행 환경에서 갱신 처리기를 받아야 갱신할 수 있습니다. 만료 시각이나 갱신 처리기가 없는 인증 정보가 자동으로 무기한 연장된다고 가정하지 않습니다.

S3의 AWS 인증 정보

S3의 세션 인증 정보는 AwsCredentials이며 access_key_id, secret_access_key, 선택적인 session_token과 expiration을 포함합니다. 다음 함수는 실제 소스 인증 정보로 S3 객체 목록의 첫 페이지를 반환합니다. 전체 목록을 얻으려면 반환된 NextContinuationToken에 따른 후속 조회가 별도로 필요합니다.

import boto3
from external_systems.sources import AwsCredentials, Refreshable, Source, SourceCredentials


def list_s3_first_page(source: Source, region: str, bucket: str) -> dict:
    provider: Refreshable[SourceCredentials] = source.get_session_credentials()
    credentials = provider.get()
    if not isinstance(credentials, AwsCredentials):
        raise TypeError("S3 소스에는 AwsCredentials가 필요합니다")

    client = boto3.client(
        "s3",
        region_name=region,
        aws_access_key_id=credentials.access_key_id,
        aws_secret_access_key=credentials.secret_access_key,
        aws_session_token=credentials.session_token,
    )
    try:
        return client.list_objects_v2(Bucket=bucket)
    finally:
        client.close()

갱신 제공자에서 새 인증 정보를 받아도 이미 정적 키로 생성한 Boto3 클라이언트가 자동으로 바뀌는 것은 아닙니다. 장시간 작업은 유효한 인증 정보를 받아 클라이언트에 반영해야 합니다. 생성한 클라이언트 객체를 Spark 실행기 클로저에 전달하는 대신, 실제로 요청을 수행하는 프로세스에서 클라이언트를 생성합니다.

OIDC 및 OAuth 인증 정보

OauthCredentials에는 access_token과 만료 시각 expiration이 있습니다. 다음 함수는 실제 자격 증명을 반환하므로 호출자가 두 값을 사용할 수 있습니다.

from external_systems.sources import OauthCredentials, Refreshable, Source, SourceCredentials


def current_oauth_credentials(source: Source) -> OauthCredentials:
    provider: Refreshable[SourceCredentials] = source.get_session_credentials()
    credentials = provider.get()
    if not isinstance(credentials, OauthCredentials):
        raise TypeError("이 소스에는 OauthCredentials가 필요합니다")
    return credentials

OIDC 신원 토큰을 원천이 직접 받는지, 원천의 토큰으로 교환해야 하는지 구분합니다. Snowflake는 OAuth 토큰으로 직접 사용할 수 있지만 Databricks는 토큰 교환이 필요합니다. 장시간 루프에서는 갱신 가능한 제공자를 보관하고 필요할 때 get()으로 현재 값을 받습니다. 만료되거나 갱신에 실패한 토큰을 그대로 재사용하지 않습니다.

에이전트 프록시를 통한 소켓 연결

Foundry 워커에 에이전트 프록시 정책을 연결하면 사설 네트워크의 시스템에 접근할 수 있습니다. HTTPS가 아닌 클라이언트에는 소스의 create_socket(host, port)가 제공하는 연결을 전달할 수 있습니다. 소켓은 사용 후 닫아야 합니다.

다음 SFTP 예제는 Fabric과 Paramiko를 사용합니다. 인수에는 실제 호스트, 포트, 신뢰할 서버 키를 담은 known_hosts 파일의 경로를 전달합니다. 소스에는 username, password 키가 구성되어 있어야 합니다. 다른 비밀 정보 이름을 사용한다면 실제 소스의 키에 맞춥니다.

import fabric
import paramiko
from external_systems.sources import Source


def list_sftp_files(
    source: Source,
    host: str,
    port: int,
    known_hosts_path: str,
    remote_directory: str = ".",
) -> list[str]:
    with source.create_socket(host, port) as proxy_socket:
        connection = fabric.Connection(
            host,
            user=source.get_secret("username"),
            port=port,
            connect_kwargs={
                "password": source.get_secret("password"),
                "sock": proxy_socket,
                "look_for_keys": False,
                "allow_agent": False,
            },
        )
        connection.client.load_host_keys(known_hosts_path)
        connection.client.set_missing_host_key_policy(paramiko.RejectPolicy())
        with connection:
            return connection.sftp().listdir(remote_directory)

서버 키 파일은 실제 서버 관리자 등 신뢰할 수 있는 경로로 확보한 키를 사용합니다. 이 예제는 알 수 없는 키를 자동 수락하지 않습니다. 프록시 연결의 인증과 SFTP 서버 키 검증은 서로 다른 단계입니다.

다른 HTTP 클라이언트의 프록시 URI

get_https_proxy_uri()는 인증 정보가 포함된 프록시 URI를 반환합니다. 프록시가 필요 없으면 None일 수 있습니다. URI에 프록시 인증 토큰이 들어갈 수 있으므로 로그에 남기지 않습니다.

다음은 기본 신뢰 저장소로 서버를 검증할 수 있고 별도 클라이언트 인증서가 필요하지 않은 HTTPX 사용 예입니다.

import httpx
from external_systems.sources import Source


def get_source_text_with_httpx(source: Source, relative_path: str) -> str:
    proxy_uri = source.get_https_proxy_uri()
    source_url = source.get_https_connection().url
    with httpx.Client(proxy=proxy_uri) as client:
        response = client.get(source_url + relative_path, timeout=10.0)
        response.raise_for_status()
        return response.text

프록시 URI만 복사했다고 소스의 서버 또는 클라이언트 인증서 설정까지 다른 라이브러리에 자동 적용되는 것은 아닙니다. 사용자 지정 CA가 필요하면 source.server_certificates_bundle_path로 제공되는 CA 번들 경로를 해당 클라이언트의 TLS 설정에 반영할 수 있습니다. 클라이언트 인증서도 요구하는 연결은 해당 라이브러리에 필요한 인증서 구성을 제공하거나 소스의 기본 HTTPS 클라이언트를 사용합니다.

소스 구성 속성

Foundry API가 구성을 공개하는 소스 유형에서는 source_configuration으로 구성 속성에 접근할 수 있습니다. 이 속성은 지원되지 않는 소스에서 ValueError를 발생시킵니다. 모든 소스에 공통 hostname 속성이 있다고 가정하지 않습니다.

Snowflake 소스의 계정 식별자를 읽는 예제는 다음과 같습니다.

from external_systems.sources import Source


def snowflake_account_identifier(source: Source) -> str:
    return source.source_configuration["accountIdentifier"]

이 함수는 API로 구성을 제공하는 실제 Snowflake 소스를 입력으로 받습니다. 구성 전체나 비밀 정보를 무분별하게 로그에 출력하지 않고 필요한 속성만 사용합니다.