---
title: "Transform data from sources using the Python SDK"
description: "We have prepared detailed guides and sample scripts for common use cases to help you get started quickly and efficiently"
url: https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/tutorials-python-sdk-transform
lang: pl
lastUpdated: 2026-09-14
---
> For AI agents: the complete documentation index is available at https://docs.ovhcloud.com/pl/llms.txt, the full documentation bundle is available at https://docs.ovhcloud.com/pl/llms-full.txt.

# Transform data from sources using the Python SDK

## Objective

We have prepared detailed guides and sample scripts for common use cases to help you get started quickly and efficiently. These examples demonstrate how to leverage **Custom Actions** in different scenarios, allowing you to extract, transform, and load (ETL) data across various components of your data platform.

## 1. Custom Action with a Lakehouse Manager Table

### Overview:

This short example shows how to extract data from a table in Lakehouse Manager, transform it and then insert it or update the Lakehouse Manager.

The code is written in [Custom Action](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/dpe-actions-custom.md) context and it uses the _stations\_rides_ table from the [Getting Started](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/landing-page-getting-started.md) Tutorial. If you did that tutorial, you can just copy and paste the code below to test it, otherwise you need to adapt it to your tables and data sources.

### Example Application:

:::warning
Do not forget to build the table you use in the Lakehouse Manager and then load it with a [DPE Load Action](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/dpe-actions-load.md). You need to load your table before using the code below, otherwise it will not work.
:::

```python
import sys
import pandas as pd
import logging

from forepaas.dwh import connect
from forepaas.dwh import bulk_insert

logger = logging.getLogger(__name__)
def customfunc(event):
    try:
        logger.notice("Begin function")

        # make connection to the default dataset
        cn = connect("dwh/default_dataset/")

        # option 1 : extract data from the table with no SQL required
        df = cn.select("stations_rides")

        # option 2 : extract data with custom SQL
        df = cn.query("SELECT station_id, date, rides, station_name FROM stations_rides")

        # perform your custom transform in the dataframe
        df.loc[df["station_name"] == 'Harlem-Lake', "rides"] = 0

        # reinsert your dataframe in the destination table
        stats = bulk_insert(cn, "stations_rides", df) 

        # show insertion statistics (if DBMS compatible) 
        logger.info(stats)

        # delete rows where station name is "Davis"
        cn.delete("stations_rides", {"station_name":"Davis"})

        # update rows set rides to 0 where station_id=40040
        cn.update("stations_rides", {"rides":0}, {"station_id":40040})

        # when finished, disconnect cn
        del cn    
        logger.notice("END function")
    except Exception as err: 
        raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))
```

## 2. Custom Action with Data Platform Buckets

### Overview:

Sometimes you need to handle a complex file format beyond our [Load Action](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/dpe-actions-load.md) capabilities.
In this case, we advise you to store and manipulate files with the [Data Platform Buckets](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/lakehouse-manager-buckets.md) in your Project.

:::info
In the current Data Platform SDK, the **Datastore connector** is used to interact with the Data Platform Buckets. You may think of the Datastore simply as a bucket container.
:::

### Example Applications:

```python
import sys
import pandas as pd
from logging import getLogger

from forepaas.dwh import connect
from forepaas.dwh import bulk_insert

logger = getLogger(__name__)
def extract_func(event):
    try:
        # we get data from a bucket and we will archive them in another bucket
        bucket_source_name = "your_source_bucket_name_here"
        bucket_archives_name = "your_source_bucket_name_here"

        # create a connector to handle bucket   
        bucket_connector = connect("data_store/{}".format(bucket_source_name))

        # list files from bucket
        files = bucket_connector.list()

        # retrieve a file from Data Store bucket to temporary local folder
        bucket_filepath = "stations_rides.csv"
        local_filepath = "/tmp/stations_rides.csv"
        bucket_connector.fget(bucket_filepath, local_filepath)

        # read then transform the file as you need
        # here the date column format is simply adjusted for compatibility reasons 
        df = pd.read_csv(local_filepath, sep=';')
        df['date'] = pd.to_datetime(df['date'])

        # load the dataframe into a project table named 'raw_file'
        cn = connect("dwh/default_dataset/")
        bulk_insert(cn, "stations_rides_artur", df)
        del cn
        
        # option 1 : copy the file into the archives bucket
        bucket_archive_filepath = "archives/stations_rides.csv"
        bucket_connector.fcopy_to(bucket_archives_name, bucket_archive_filepath, bucket_filepath)
       
        # option 2 : put a file into the archives
        bucket_archives = connect("data_store/{}".format(bucket_archives_name))
        bucket_archives.fput(bucket_archive_filepath, local_filepath)
        del bucket_archives

        # delete file from source bucket
        bucket_connector.delete(bucket_filepath)

        # disconnect from datastore
        del bucket_connector
    except Exception as err: 
        raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))
```

Below is an example code which uploads an image to a bucket from a simple URL.

```python
from forepaas.dwh import connect

data_store = connect('data_store')

# Get bucket and upload image from URL to path uploads/test.jpg.
# And finally get the image from the bucket
bucket_test = data_store.get_bucket('test')

lists = bucket_test.list(recursive=True)

bucket_test.put_request("https://i.stack.imgur.com/r8jTK.jpg", path='uploads/test.jpg')
data = bucket.get('hello/test.jpg')

# Create a bucket if it does not already exists
if data_store.bucket_exists('test-exists') is False:
    data_store.create_bucket('test-exists')

# Connect directly to the bucket test and remove the file
bucket_test2 = connect('data_store/test')
bucket_test2.delete('hello/test.jpg')
```

:::tip
This also works with any **Object Store** that you may define as **S3<sup>1</sup> compatible source** in the Connectors.
:::

## 3. Custom Action with a Connectors Source

### Overview:

This example will show you how to get a file directly from a [Connectors Source](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/landing-page-connectors-sources.md), treat it and put the treated data into a [Lakehouse Manager Table](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/lakehouse-manager-tables.md).

The code is written in the [Custom Action](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/dpe-actions-custom.md) context and it uses the _chicago\_files_ source from the [Getting Started Tutorial](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/landing-page-getting-started.md). If you did that tutorial, you can just copy and paste the code below to test it, otherwise you need to adapt it to your tables and data sources.

:::tip
This works with any `Source` protocol such as [FTP](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/connectors-sources-ftp.md), [Dropbox](https://docs.ovhcloud.com/pl/guides/public-cloud/data-platform/connectors-sources-dropbox.md), etc.
:::

### Example Applications:

```python
import sys
from forepaas.dwh.connect import connect
from forepaas.dwh import bulk_insert
import logging

logger = logging.getLogger(__name__)

def customfunc(event):
    try:
        # here we are connecting to a source named 'chicago_files'
        source_address = "dwh/chicago_files_artur/"

        # specify unsupported filename from list of files in source
        filename_w_extension = "stations_rides.csv"

        # connecto to file toget the address 
        source_file_connector = connect(source_address + filename_w_extension) 

        # connect to file directly
        file_address = source_file_connector.get()
        file_connector = connect(file_address) 

        # Extract and treat the file so it is usable
        df = file_connector.extract(return_type='dataframe') 

        # Treat the data
        # - - - - 

        # connect to the Lakehouse Manager
        dm_connector = connect("dwh/default_dataset/")

        # insert into an existing destination table
        stats = bulk_insert(dm_connector, "chicago_calendar_full", df) 

        logger.info(stats)

        # disconnect from datastore and remote source
        del source_connector
        del dm_connector

    except Exception as err:
            raise Exception(f"err:{err} L:{sys.exc_info()[2].tb_lineno}")
```

## Go further

If you need training or technical assistance to implement our solutions, contact your sales representative or click on [this link](https://www.ovhcloud.com/pl/professional-services/) to get a quote and ask our Professional Services experts for a custom analysis of your project.

Ask questions, give your feedback and interact directly with the team building the Data Platform on the dedicated [Discord channel](https://discord.gg/ovhcloud).

If you need support with your OVHcloud services, create a request in our [Help Centre](https://help.ovhcloud.com/csm?id=csm_get_help).

Join our [community of users](https://community.ovhcloud.com/).

1
: S3 is a trademark of Amazon Technologies, Inc. OVHcloud's service is not sponsored by, endorsed by, or otherwise affiliated with Amazon Technologies, Inc.