Sources in Python environments

Foundry provides the ability to connect to external systems in Python environments across the platform. These capabilities include source-based external transforms, external functions, and compute modules ↗. This page discusses common use cases and workflows for external systems in Python environments. For more information, visit Palantir's external-systems ↗ open source library.

Source initialization is not included in any of the examples below, as this will vary between environments. To learn how to obtain an initialized source object, refer to the environment's (for example, a transforms repository, functions repository, compute module, etc.) relevant documentation on usage of sources. You may additionally find snippets in the source information panel side bar with instructions for usage.

HTTPS client

For REST-based sources, Palantir provides a preconfigured HTTPS client built on top of the Python requests library. For on-premises systems using agent proxy, the client will come preconfigured with all necessary proxy configurations.

Note that sources must be initialized with only one HTTP connection; sources initialized with more or less than one HTTP connection are considered invalid and a preconfigured client will not be created. If you attempt to create a connection with an invalid source connection configuration, you will receive an error Only single connection sources are supported.

To find out how many connections your source has, refer to the source's sidebar panel in the External connection section of your given environment, as seen in the example below:

Source connections displayed on the Foundry code resource sidebar

Copied!1
2
3
4
5
6
7
8
9
10
11
12
from external_systems.sources import Source, HttpsConnection
from requests import Session

my_source: Source = # Source is initialized differently based on the environment

https_connection: HttpsConnection = my_source.get_https_connection()

external_system_url: str = https_connection.url

http_client: Session = https_connection.get_client()

response = http_client.get(external_system_url + "/api/v1/example/", timeout=10)

Secrets

Source secrets can be referenced using get_secret("<secret_name>") on the source.

Copied!1
2
3
4
5
from external_systems.sources import Source

my_source: Source = ...

my_secret: str = my_source.get_secret("SECRET_NAME")

Credentials generation and refresh management are available for sources using session credentials. For any S3 source authenticated using OIDC or Cloud Identity, you can generate session credentials using get_aws_credentials(). If the source is not an S3 source or if the source is incorrectly configured, this method will throw an error.

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import boto3

from external_systems.sources import Source, Refreshable, AwsCredentials

S3_BUCKET_REGION = <aws_region>
S3_BUCKET_NAME = <bucket_name>

s3_source: Source = ...

refreshable_credentials: Refreshable[AwsCredentials] = s3_source.get_aws_credentials()

session_credentials: AwsCredentials = refreshable_credentials.get()

s3_client = boto3.client(
    "s3",
    region_name=S3_BUCKET_REGION,
    aws_access_key_id=session_credentials.access_key_id,
    aws_secret_access_key=session_credentials.secret_access_key,
    aws_session_token=session_credentials.session_token,
)

s3_response = s3_client.list_objects_v2(Bucket=S3_BUCKET_NAME)

On-premises connectivity with Foundry agent proxy

The Foundry agent proxy allows connections in code to be established to on-premise systems as if the connections were over the open internet. For more details on how the Foundry agent proxy is configured, refer to the agent proxy runtime documentation.

Socket connections

For non-HTTPS connections to external systems that require connections through Foundry's agent proxy, a preconfigured socket is provided. Below is an example of using this socket with an on-premise SFTP server connection.

On-premise SFTP server example

This example uses the Fabric ↗ library.

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import fabric

from external_systems.sources import Source
from socket import socket

SFTP_HOST = <sftp_host>
SFTP_PORT = <sftp_port>

on_prem_sftp_server_source: Source = ...

username: str = on_prem_sftp_server_source.get_secret("username")
password: str = on_prem_sftp_server_source.get_secret("password")

proxy_socket: socket = on_prem_sftp_server_source.create_socket(SFTP_HOST, SFTP_PORT)

with fabric.Connection(
    SFTP_HOST,
    user=username,
    port=SFTP_PORT,
    connect_kwargs={
        "password": password,
        "sock": proxy_socket,
    },
) as sftp_conn:
    sftp_client = sftp_conn.sftp()
    file_list = sftp_client.listdir(".")

Authenticated proxy URI

For more granular use cases, a pre-authenticated proxy URI is provided to allow connections to on-premises external systems.

Non-requests library example

In cases where the requests library client is not sufficient, you may need to use another HTTP client. This example uses the HTTPX ↗ library.

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
import httpx

from external_systems.sources import Source
from typing import Optional

agent_proxy_source: Source = ...

authenticated_proxy_uri: Optional[str] = agent_proxy_source.get_https_proxy_uri()

source_url: str = agent_proxy_source.get_https_connection().url

with httpx.Client(proxy=authenticated_proxy_uri) as client:
    response = client.get(source_url + "/api/v1/example/", timeout=10.0)

Source properties

For source types that are available via the Foundry API, the configuration properties can also be directly accessed for use in code.

Copied!1
2
3
4
5
from external_systems.sources import Source

snowflake_source: Source = ...

account_id: str = snowflake_source.source_configuration.get("accountIdentifier")

←

PREVIOUSExternal functions

NEXTLegacy external transforms

→