### Install Compute Module Library Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Installs the foundry-compute-modules package using pip. Includes instructions for installing optional dependencies for Foundry Sources support. ```bash pip install foundry-compute-modules pip install foundry-compute-modules[sources] ``` -------------------------------- ### Register Multiple Functions and Start Compute Module Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Illustrates registering multiple functions (add, hello) explicitly using `add_functions` and then starting the compute module service. ```python def add(context, event) -> int: return event["x"] + event["y"] def hello(context, event) -> str: return "Hello " + event["x"] + "!" from compute_modules import add_functions, start_compute_module if __name__ == "__main__": add_functions( hello, add, ) start_compute_module() ``` -------------------------------- ### Install Python Packages and Scripts Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Installs all dependencies specified in pyproject.toml and makes scripts available in the Python environment. Requires poetry. ```sh poetry install --extras sources ``` -------------------------------- ### Get Foundry Source Client Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Retrieves a pre-configured client for a Foundry Source imported into a Compute Module. Requires prior installation of dependencies as per the installation guide. ```Python from compute_modules.sources_v2 import get_source from external_systems.sources import Source source: Source = get_source("") ``` -------------------------------- ### Build Python Library Locally Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Builds the Python library, producing a tar.gz and a .whl file in the ./dist directory. These files can be installed locally for testing. ```sh poetry build ``` ```sh % poetry build Building foundry-compute-modules (0.0.0) - Building sdist - Built foundry_compute_modules-0.0.0.tar.gz - Building wheel - Built foundry_compute_modules-0.0.0-py3-none-any.whl % pip install ./dist/foundry_compute_modules-0.0.0.tar.gz ``` -------------------------------- ### Avoid dict/Dict with No Type Parameters Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Example of an anti-pattern: using generic dict or typing.Dict without specifying the key and value types. This will result in exceptions. ```Python # These both will raise an exception @dataclass class MyPayload: data: dict @dataclass class MyPayload: data: typing.Dict ``` -------------------------------- ### Avoid Python Class with No Constructor Type Hints Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Example of an anti-pattern: a Python class used as an input type where the constructor lacks type hints for its parameters. This will cause exceptions. ```Python # This will raise an exception class BadClassNoInitHints: arg1: str arg2: int def __init__(self, arg1, arg2): ... ``` -------------------------------- ### Avoid Python Class with No Class Type Hints Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Example of an anti-pattern: a Python class used as an input type without type hints on its attributes. This will lead to exceptions. ```Python # This will raise an exception class BadClassNoTypeHints: def __init__(self, arg1: str, arg2: int): ... ``` -------------------------------- ### Retrieve Third-Party App Credentials and Token Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Facilitates obtaining client ID and credentials for Application's permissions, enabling a service user for authentication. Also shows how to get a scoped OAuth token. ```Python from compute_modules.auth import retrieve_third_party_id_and_creds, oauth CLIENT_ID, CLIENT_CREDS = retrieve_third_party_id_and_creds() # get a scoped token for your 3pa HISTORY = "myenvironment.palantirfoundry.com" access_token = oauth(HOSTNAME, ["api:datasets-read"]) ``` -------------------------------- ### Avoid Python Class with **kwargs in Constructor Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Example of an anti-pattern: a Python class used as an input type where the constructor accepts variable keyword arguments (**kwargs). This is not supported and will raise an exception. ```Python # This will raise an exception class BadClassKwargsInit: arg1: str arg2: int def __init__(self, arg1: str, arg2: int, **kwargs): ... ``` -------------------------------- ### Retrieve Compute Module Arguments Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Provides utilities to get arguments passed to a compute module. get_raw_arguments returns a list of strings, while get_parsed_arguments returns a parsed namespace object. ```Python # app.py import logging as log from compute_modules.annotations import function from compute_modules.arguments import get_raw_arguments, get_parsed_arguments log.basicConfig(level=log.INFO) @function def hello(context, event) -> str: raw_args = get_raw_arguments() parsed_args = get_parsed_arguments() log.info(f"raw_args: {raw_args}") log.info(f"parsed_args: {parsed_args}") ... ``` -------------------------------- ### Avoid Python Class with *args in Constructor Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Example of an anti-pattern: a Python class used as an input type where the constructor accepts variable positional arguments (*args). This is not supported and will raise an exception. ```Python # This will raise an exception class BadClassArgsInit: arg1: str arg2: int def __init__(self, arg1: str, arg2: int, *args): ... ``` -------------------------------- ### Apply Custom JSON Log Formatter with Compute Module SDK Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Demonstrates how to use the `setup_logger_formatter` utility to apply a custom JSON formatter to the compute module's logging. This allows capturing compute module-specific details like process and job IDs in JSON format. ```Python import logging from compute_modules.logging import setup_logger_formatter import json # Write our a custom formatter that makes a JSON log string line class JsonFormatter(logging.Formatter): def format(self, record: Any) -> str: log_record = { "level": record.levelname, "process_id": record.process_id, "job_id": record.job_id, "location": f"{record.filename}:{record.lineno}", "message": record.getMessage(), } return json.dumps(log_record) setup_logger_formatter(JsonFormatter()) ``` -------------------------------- ### Run Tests Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Executes the test suite for the Python compute module. ```sh poe test ``` -------------------------------- ### Class with Class and Constructor Type Hints Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Illustrates defining an input type using a regular Python class with type hints for both class attributes and constructor parameters. This is a valid approach for defining structured inputs. ```Python from compute_modules.annotations import function class GoodExample: some_flag: bool some_value: int def __init__(self, some_flag: bool, some_value: int) -> None: self.some_flag = some_flag self.some_value = some_value @function def typed_function(context, event: GoodExample) -> int: return event.some_value ``` -------------------------------- ### Run Formatter (Black, Ruff, ISort) Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Applies code formatting using Black, Ruff, and ISort to automatically fix identified issues. ```sh poe format ``` -------------------------------- ### Annotating QueryContext for Static Typing Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Demonstrates how to annotate the `context` parameter of a compute module function with `QueryContext` for static type checking, providing access to job-specific information. ```Python from typing import TypedDict from compute_modules.context import QueryContext from compute_modules.annotations import function class HelloInput(TypedDict): x: str @function def hello(context: QueryContext, event: HelloInput) -> str: return f"Hello {event['x']}! Your job ID is: {context.jobId}" ``` -------------------------------- ### Compute Module Logging Utility Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Recommends using the get_logger utility function for emitting logs. It returns a standard logging.Logger instance, allowing for drop-in replacement of logging.getLogger. ```Python from compute_modules.logging import get_logger logger = get_logger(__name__) logger.setLevel(logging.INFO) logger.debug("Can't see me") logger.info("Peekaboo!") logger.warning("Peekaboo!") logger.error("Peekaboo!") logger.critical("Peekaboo!") ``` -------------------------------- ### Dataclass Input Type with Various Type Hints Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Demonstrates using a dataclass as an input type for a compute module function, including various Python built-in types and standard library types like datetime and decimal. ```Python from dataclasses import dataclass from compute_modules.annotations import function import datetime import decimal @dataclass class TypedInput: bytes_value: bytes bool_value: bool date_value: datetime.date decimal_value: decimal.Decimal float_value: float int_value: int str_value: str datetime_value: datetime.datetime other_date_value: datetime.datetime @function def typed_function(context, event: TypedInput) -> str: diff = event.other_date_value - event.datetime_value return f"The diff between dates provided is {diff}" ``` -------------------------------- ### Run Linter Checks (Black, Ruff, ISort) Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Runs code formatting and linting checks using Black, Ruff, and ISort. This command identifies issues but does not fix them. ```sh poe check_format ``` -------------------------------- ### Retrieve Pipeline Resources Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Fetches configured resources for a pipeline module, allowing access to their RIDs and branches via aliases. ```Python from compute_modules.resources import PipelineResource, get_pipeline_resources resources: dict[str, PipelineResource] = get_pipeline_resources() print(f"My resource's rid is: {resources['your-alias-name'].rid}") ``` -------------------------------- ### Retrieve Source Information (Deprecated) Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Shows a deprecated method for retrieving source information (secrets and configurations) using the context object within a compute module function. ```python from compute_modules.annotations import function @function def get_sources(context, event) -> dict: source_secrets = context["sources"] source_configs = context["source_configs"] return { "secrets": source_secrets, "configs": source_configs } ``` -------------------------------- ### Register Function with @function Annotation Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Demonstrates how to use the @function annotation to register a single compute module function. The function takes a context and an event, returning an integer. ```python from compute_modules.annotations import function @function def add(context, event) -> int: return event["x"] + event["y"] ``` -------------------------------- ### Stream Function Results with add_function Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Shows how to enable streaming results for a function registered with `add_function` by passing `streaming=True`. ```python def get_strings(context, event) -> list[str]: return [f'string {i}' for i in range(10)] from compute_modules import add_functions, start_compute_module if __name__ == "__main__": add_function(get_strings, streaming=True) start_compute_module() ``` -------------------------------- ### Run Mypy Checks Source: https://github.com/palantir/python-compute-module/blob/develop/DEV_SETUP.md Performs static type checking using mypy to identify potential type errors in the code. ```sh poe check_mypy ``` -------------------------------- ### Deprecated Source Information Retrieval Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Provides code snippets for accessing source secrets and configurations using deprecated APIs. It is recommended to use the newer 'Sources in Compute Modules' section for current best practices. ```Python from compute_modules.sources import get_sources, get_source_secret, get_source_configurations, get_source_config # retrieve a dict with all sources all_source_creds = get_sources() all_source_configs = get_source_configurations() # retrieve the credentials of a specific source my_creds = get_source_secret("", "MyCredential") # retrieve the source configuration my_config = get_source_config("") ``` -------------------------------- ### Retrieve Pipeline Token Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Obtains an authentication token for interacting with Foundry resources in Pipeline mode. Used with libraries like 'requests' for authenticated calls. ```Python from compute_modules.auth import retrieve_pipeline_token import requests pipeline_token = retrieve_pipeline_token() requests.post(..., headers={"Authorization": f"Bearer {pipeline_token}") ``` -------------------------------- ### Refreshing OAuth Token Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Utilizes the RefreshingOauthToken class to automatically refresh OAuth tokens after expiry, typically set to 30 minutes. Tokens should only be retrieved via the get_token() function. ```Python from compute_modules.auth import RefreshingOauthToken refreshing_token = RefreshingOauthToken(hostname=HOSTNAME, scope=["api:datasets-read"]) # Token will automatically refresh when beyond expiry period access_token = refreshing_token.get_token() ``` -------------------------------- ### TypedDict Input Type Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Defines an input type for a compute module function using Python's TypedDict. This allows for static type checking of the input event. ```Python from typing import TypedDict from compute_modules.annotations import function class HelloInput(TypedDict): planet: str @function def hello(context, event: HelloInput) -> str: return "Hello " + event["planet"] + "!" ``` -------------------------------- ### Stream Function Results with @function Annotation Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Demonstrates enabling streaming results for a function that returns an iterable. The `@function(streaming=True)` annotation is used. ```python from compute_modules.annotations import function @function(streaming=True) def get_strings(context, event) -> list[str]: return [f'string {i}' for i in range(10)] ``` -------------------------------- ### Set Compute Module Internal Log Level Source: https://github.com/palantir/python-compute-module/blob/develop/README.md Explains how to adjust the internal logging level of the `compute_modules` library using the `set_internal_log_level` function. This is useful for debugging by allowing lower-level logs (e.g., DEBUG, INFO) to be surfaced. ```Python from compute_modules.logging import set_internal_log_level import logging set_internal_log_level(logging.DEBUG) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.