## Build relational assets

URL: https://docs.atlan.com/product/capabilities/build-apps/sdks/python/packages/how-tos/build-relational-assets

> Create and update relational assets (connections, databases, schemas, tables, views, columns) in Atlan using RelationalAssetsBuilder and the Python SDK (pyatlan).

# RelationalAssetsBuilder: build relational assets

Use `RelationalAssetsBuilder` in the Atlan Python SDK to programmatically create and update net-new relational assets.

The [relational assets builder package](https://docs.atlan.com/llms/catalog/discovery/relational-assets-builder/llms.txt)
allows you to create (and update) net-new relational assets: connections, databases, schemas, tables, views, materialized views and columns.

## Import relational assets from object store

To import relational assets directly from the object store:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Import relational assets from the object store"
from pyatlan.client.atlan import AtlanClient
from pyatlan.model.packages import RelationalAssetsBuilder
from pyatlan.model.assets import Asset
from pyatlan.model.enums import AssetInputHandling, AssetDeltaHandling, AssetRemovalType

client = AtlanClient()

workflow = (
 RelationalAssetsBuilder() # (1)
 .object_store( # (2)
 prefix="/test/prefix",
 object_key="assets-test.csv",
 )
 .s3( # (3)
 access_key="test-access-key",
 secret_key="test-secret-key",
 bucket="my-bucket",
 region="us-west-1",
 )
 .assets_semantics( # (4)
 input_handling=AssetInputHandling.UPSERT,
 delta_handling=AssetDeltaHandling.INCREMENTAL,
 removal_type=AssetRemovalType.ARCHIVE,
 )
 .options( # (5)
 remove_attributes=[Asset.CERTIFICATE_STATUS, Asset.ANNOUNCEMENT_TYPE],
 fail_on_errors=True,
 field_separator=",",
 batch_size=20,
 )
).to_workflow() # (6)

response = client.workflow.run(workflow) # (7)
```

1. The `RelationalAssetsBuilder` allows you to create (and update) net-new relational assets.
2. To set up the package for importing metadata directly from the object store, provide the following information:

 - `prefix`: directory (path) within the bucket/container
 from which to retrieve the objects.
 - `object_key`: object key (filename), including its extension,
 within the bucket/container and prefix.
3. You can use different object store methods (e.g: `s3()`, `gcs()`, `adls()`). In this example,
we're building a workflow using `s3()` and for that, you’ll need to provide the following information:

 - AWS access key.
 - AWS secret key.
 - name of the bucket/storage that contains the metadata CSV files.
 - name of the AWS region.
4. To set up the package to import metadata with semantics, you need to provide:

 - `input_handling`: whether to allow the creation of new full (`AssetInputHandling.UPSERT`)
 or partial (`AssetInputHandling.PARTIAL`) assets from the input CSV, or make sure assets
 are only updated (`AssetInputHandling.UPDATED`) if they already exist in Atlan.
 - `delta_handling`: whether to treat the input file as an initial load, full replacement
 [`AssetDeltaHandling.FULL_REPLACEMENT`] (deleting any existing assets not in the file) or only incremental [`AssetDeltaHandling.INCREMENTAL`] (no deletion of existing assets).
 - `removal_type`: if `delta_handling` is set to `FULL_REPLACEMENT`, this parameter specifies whether to
 delete any assets not found in the latest file by archive (recoverable) [`AssetRemovalType.ARCHIVE`] or purge (non-recoverable) [`AssetRemovalType.PURGE`].
 If `delta_handling` is set to `INCREMENTAL`, this parameter is ignored and assets are archived.

5. (Optional) To set up the package for importing relational assets with
advanced configuration, provide the following information:

 - `remove_attributes`: list of attributes to clear (remove)
 from assets if their value is blank in the provided file.
 - `fail_on_errors`: specifies whether an invalid value
 in a field should cause the import to fail (`True`) or
 log a warning, skip that value, and proceed (`False`).
 - `field_separator`: character used to separate
 fields in the input file (e.g: `','` or `';'`).
 - `batch_size`: maximum number of rows
 to process at a time (per API request).
6. Convert the package into a `Workflow` object.
7. Run the workflow by invoking the `run()` method
on the workflow client, passing the created object.

 :::warning[Workflows run asynchronously]
Remember that workflows run asynchronously.
See the [packages and workflows introduction](https://docs.atlan.com/llms/platform/python/packages/llms.txt)
for details on how to check the status and wait
until the workflow has been completed.
 :::

### Kotlin

:::warning[Coming soon]
:::

### Raw REST API

:::tip[Create the workflow via UI only]
We recommend creating the workflow only via the UI.
To rerun an existing workflow, see the steps below.
:::

## Re-run existing workflow

To re-run an existing relational assets builder workflow:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Re-run existing relational assets builder workflow"
from pyatlan.client.atlan import AtlanClient
from pyatlan.model.enums import WorkflowPackage

client = AtlanClient()

existing = client.workflow.find_by_type( # (1)
 prefix=WorkflowPackage.RELATIONAL_ASSETS_BUILDER, max_results=5
)

# Determine which relational assets builder workflow (n)

# from the list of results you want to re-run.

response = client.workflow.rerun(existing[n]) # (2)
```

1. You can find workflows by their type using the workflow client `find_by_type()`
method and providing the **prefix** for one of the packages.
In this example, we do so for the `RelationalAssetsBuilder`. (You can also specify
the **maximum number of resulting workflows** you want to retrieve as results.)
2. Once you've found the workflow you want to re-run,
you can simply call the workflow client `rerun()` method.

 - Optionally, you can use `rerun(idempotent=True)` to avoid re-running a workflow that's already in running or in a pending state.
 This will return details of the already running workflow if found, and by default, it's set to `False`.

 :::warning[Workflows run asynchronously]
Remember that workflows run asynchronously. See the [packages and workflows introduction](https://docs.atlan.com/llms/platform/python/packages/llms.txt)
for details on how you can check the status and wait until the workflow has been completed.
 :::

### Kotlin

:::warning[Coming soon]
:::

### Raw REST API

:::warning[Requires multiple steps through the raw REST API]
1. Find the existing workflow.
2. Send through the resulting re-run request.
:::
```json showLineNumbers title="POST /api/service/workflows/indexsearch"
{
 "from": 0,
 "size": 5,
 "query": {
 "bool": {
 "filter": [
 {
 "nested": {
 "path": "metadata",
 "query": {
 "prefix": {
 "metadata.name.keyword": {
 "value": "csa-relational-assets-builder" // (1)
 }
 }
 }
 }
 }
 ]
 }
 },
 "sort": [
 {
 "metadata.creationTimestamp": {
 "nested": {
 "path": "metadata"
 },
 "order": "desc"
 }
 }
 ],
 "track_total_hits": true
}
```

1. Searching by the `csa-relational-assets-builder` prefix will make sure you only find existing relational assets builder workflows.

 :::tip[Name of the workflow]
The name of the workflow will be nested within the `_source.metadata.name` property of the response object.
(Remember since this is a search, there could be multiple results, so you may want to use the other
details in each result to determine which workflow you really want.)
 :::
```json title="POST /api/service/workflows/submit"
{
 "namespace": "default",
 "resourceKind": "WorkflowTemplate",
 "resourceName": "csa-relational-assets-builder-1684500411" // (1)
}
```

2. Send the name of the workflow as the `resourceName` to rerun it.

---