Transformer les données des sources avec le SDK 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")

        # connexion au dataset par défaut
        cn = connect("dwh/default_dataset/")

        # option 1 : extraire les données de la table sans SQL requis
        df = cn.select("stations_rides")

        # option 2 : extraire les données avec une requête SQL personnalisée
        df = cn.query("SELECT station_id, date, rides, station_name FROM stations_rides")

        # effectuez votre transformation personnalisée dans le dataframe
        df.loc[df["station_name"] == 'Harlem-Lake', "rides"] = 0

        # réinsérez votre dataframe dans la table de destination
        stats = bulk_insert(cn, "stations_rides", df) 

        # affichez les statistiques d'insertion (si compatible avec le SGBD) 
        logger.info(stats)

        # supprimez les lignes où le nom de la station est "Davis"
        cn.delete("stations_rides", {"station_name":"Davis"})

        # mettez à jour les lignes en définissant rides à 0 où station_id=40040
        cn.update("stations_rides", {"rides":0}, {"station_id":40040})

        # une fois terminé, déconnectez cn
        del cn    
        logger.notice("END function")
    except Exception as err: 
        raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))

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:
        # nous récupérons les données d'un bucket et nous les archiverons dans un autre bucket
        bucket_source_name = "your_source_bucket_name_here"
        bucket_archives_name = "your_source_bucket_name_here"

        # créez un connector pour gérer le bucket   
        bucket_connector = connect("data_store/{}".format(bucket_source_name))

        # listez les fichiers du bucket
        files = bucket_connector.list()

        # récupérez un fichier du bucket Data Store vers un dossier local temporaire
        bucket_filepath = "stations_rides.csv"
        local_filepath = "/tmp/stations_rides.csv"
        bucket_connector.fget(bucket_filepath, local_filepath)

        # lisez puis transformez le fichier selon vos besoins
        # ici le format de la colonne date est simplement ajusté pour des raisons de compatibilité 
        df = pd.read_csv(local_filepath, sep=';')
        df['date'] = pd.to_datetime(df['date'])

        # chargez le dataframe dans une table du project nommée 'raw_file'
        cn = connect("dwh/default_dataset/")
        bulk_insert(cn, "stations_rides_artur", df)
        del cn
        
        # option 1 : copiez le fichier dans le bucket d'archives
        bucket_archive_filepath = "archives/stations_rides.csv"
        bucket_connector.fcopy_to(bucket_archives_name, bucket_archive_filepath, bucket_filepath)
       
        # option 2 : déposez un fichier dans les archives
        bucket_archives = connect("data_store/{}".format(bucket_archives_name))
        bucket_archives.fput(bucket_archive_filepath, local_filepath)
        del bucket_archives

        # supprimez le fichier du bucket source
        bucket_connector.delete(bucket_filepath)

        # déconnectez du datastore
        del bucket_connector
    except Exception as err: 
        raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))

from forepaas.dwh import connect

data_store = connect('data_store')

# Récupérez le bucket et uploadez l'image depuis l'URL vers le chemin uploads/test.jpg.
# Et enfin récupérez l'image depuis le 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')

# Créez un bucket s'il n'existe pas déjà
if data_store.bucket_exists('test-exists') is False:
    data_store.create_bucket('test-exists')

# Connectez-vous directement au bucket test et supprimez le fichier
bucket_test2 = connect('data_store/test')
bucket_test2.delete('hello/test.jpg')

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

logger = logging.getLogger(__name__)

def customfunc(event):
    try:
        # ici nous nous connectons à une source nommée 'chicago_files'
        source_address = "dwh/chicago_files_artur/"

        # spécifiez le nom de fichier non pris en charge depuis la liste des fichiers de la source
        filename_w_extension = "stations_rides.csv"

        # connectez-vous au fichier pour obtenir l'adresse 
        source_file_connector = connect(source_address + filename_w_extension) 

        # connectez-vous directement au fichier
        file_address = source_file_connector.get()
        file_connector = connect(file_address) 

        # Extrayez et traitez le fichier pour qu'il soit utilisable
        df = file_connector.extract(return_type='dataframe') 

        # Traitez les données
        # - - - - 

        # connectez-vous au Lakehouse Manager
        dm_connector = connect("dwh/default_dataset/")

        # insérez dans une table de destination existante
        stats = bulk_insert(dm_connector, "chicago_calendar_full", df) 

        logger.info(stats)

        # déconnectez du datastore et de la source distante
        del source_connector
        del dm_connector

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