# transforms.api.transform

## *class* transforms.api.transform(\*\*kwargs) {#transforms.api.transform}

Wrap a compute function as a [`Transform`](index.md#transforms.api.Transform) object.

The `transform` decorator is used to construct a [`Transform`](index.md#transforms.api.Transform) object from a compute function. The names used for inputs, outputs or other parameters should be the function arguments of the wrapped compute function. At compute-time, parameters are instantiated to specific objects defined by each parameter type.

```python
>>> @transform(
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
...     first_output=Output('/path/to/first/output/dataset'),
...     second_output=Output('/path/to/second/output/dataset'),
... )
... def my_compute_function(first_input, second_input, first_output, second_output):
...     # type: (TransformInput, TransformInput, TransformOutput, TransformOutput) -> None
...     first_output.write_dataframe(first_input.dataframe())
...     second_output.write_dataframe(second_input.dataframe())
```

* **Parameters:**
  **\*\*kwargs** ([*Param*](../api-param/index.md#transforms.api.Param)) – `kwargs` comprised of named [`Param`](../api-param/index.md#transforms.api.Param) or subclasses.

:::callout{theme="neutral"}
The compute function is responsible for writing data to its outputs.
:::

### Optional context parameter {#optional-context-parameter}

You can optionally accept a context object in your compute function by adding a `ctx` argument. For regular Spark transforms, the context is a [`transforms.api.TransformContext`](../api-transformcontext/index.md#transforms.api.TransformContext), which exposes runtime information and utilities such as `abort_job()`, `auth_header`, `environment`, `fallback_branches`, `parameters`, and `spark_session`.

```python
>>> @transform(
...     my_input=Input('/path/to/input'),
...     my_output=Output('/path/to/output'),
... )
... def my_compute_function(ctx, my_input, my_output):
...     # ctx is a TransformContext
...     df = my_input.dataframe()
...     # Optionally use ctx, for example, to access parameters or abort the job
...     if ctx.parameters.get('abort'):
...         ctx.abort_job()
...     my_output.write_dataframe(df)
```

For lightweight transforms created with `transform.using(...)`, the injected context is a [`transforms.api.LightweightContext`](../api-lightweightcontext/index.md#transforms.api.LightweightContext), which exposes methods and properties such as `abort_job()`, `auth_header`, and `is_incremental`.

```python
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... )
... def compute_func(ctx, my_input, my_output):
...     if not ctx.is_incremental:
...         # End execution and abort all output transactions
...         ctx.abort_job()
...     my_output.write_pandas(my_input.pandas())
```

### Incremental execution {#incremental-execution}

To run a transform incrementally, wrap it with the [`incremental`](../api-incremental/index.md#transforms.api.incremental) decorator. The incremental decorator uses output build history to determine the previous state and converts `TransformInput`, `TransformOutput`, and `TransformContext` to their incremental counterparts.

```python
# Spark transform, executed incrementally
@incremental()
@transform(
    first_input=Input('/path/to/first/input/dataset'),
    first_output=Output('/path/to/first/output/dataset'),
)
def my_compute_function(first_input, first_output):
    first_output.write_dataframe(first_input.dataframe())
```

```python
# Lightweight transform, executed incrementally
@incremental()
@transform.using(
    my_input=Input('/input'),
    my_output=Output('/output')
)
def compute_func(my_input, my_output):
    my_output.write_pandas(my_input.pandas())
```

:::callout{theme="warning"}
If your transform performs complex logic such as joins, aggregations, or distinct operations, review the [incremental overview](../../../transforms-python/incremental-overview/index.md) before using the decorator.
:::

### *classmethod* using(\*\*kwargs) {#transforms.api.transform.using}

Construct a lightweight transform via the transform class.

A lightweight transform is a transform that runs without Spark, on a single node. Lightweight transforms are faster and more cost-effective for small to medium-sized datasets. Lightweight transforms also provide more methods for accessing datasets; however, they only support a subset of the API of a regular transform, including pandas, Polars (via [`transform_polars`](../api-transform-polars/index.md#transforms.api.transform_polars)), and the filesystem API. Spark profiles are not supported for lightweight transforms.

This method accepts only input/output parameters. Resource requirements and container settings can be configured by chaining additional method calls.

For more information, see the [Transforms Python overview](../../../transforms-python/overview/index.md). For Polars DataFrame-based transformations, see [`transform_polars`](../api-transform-polars/index.md#transforms.api.transform_polars).

:::callout{theme="neutral"}
For file-level access, regular Spark transforms use [`transforms.api.FileSystem`](../api-filesystem/index.md#transforms.api.FileSystem), while lightweight transforms use [`transforms.api.FoundryDataSidecarFileSystem`](../api-foundrydatasidecarfilesystem/index.md#transforms.api.FoundryDataSidecarFileSystem).
:::

* **Parameters:**
  **\*\*kwargs** – Named [`LightweightInput`](../api-lightweightinput/index.md#transforms.api.LightweightInput) or [`LightweightOutput`](../api-lightweightoutput/index.md#transforms.api.LightweightOutput) objects for the transform’s inputs and outputs.
* **Returns:**
  A builder object that can be configured with additional method calls.

#### Examples

```python
>>> # Basic usage with default resources
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... )
... def compute_func(my_input, my_output):
...     my_output.write_pandas(my_input.pandas())
```

```python
>>> # Configure resources
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... ).with_resources(
...     cpu_cores=4,
...     memory_gb=8
... )
... def compute_func(my_input, my_output):
...     my_output.write_pandas(my_input.pandas())
```

```python
>>> # Configure container
>>> @transform.using(
...     my_output=Output('/output')
... ).with_container(
...     container_image='my-image',
...     container_tag='0.0.1'
... )
... def run_data_generator_executable(my_output):
...     os.system('$USER_WORKING_DIR/data_generator')
...     my_output.write_table(pd.read_csv('data.csv'))
```