Order & Retail Data Integration Guide

import boto3 import os from datetime import datetime, timezone # S3 bucket configuration BUCKET_NAME = 'workmagic-inbound-data' # Name of the S3 bucket where files will be uploaded DATA_CATEGORY = "{DATA_TYPE}" TENANT_PREFIX = f'account-{WM_ACCOUNT_ID}/{DATA_CATEGORY}' # Tenant-specific prefix used to organize files in the bucket # Initialize AWS session with credentials # Note: Replace with your actual credentials initial_session = boto3.Session(aws_access_key_id='{aws_access_key_id_of_client}', aws_secret_access_key='{aws_secret_access_key_of_client}', region_name='us-east-1') # 1. Get temporary credentials using STS # Create STS client to assume a role sts_client = initial_session.client('sts') try: # Assume IAM role with appropriate permissions # Replace with your actual role ARN, session name, and external ID assumed_role = sts_client.assume_role( RoleArn='arn:aws:iam::588759585297:role/{WM_ACCOUNT_ID}', RoleSessionName='{session_name}' #Take any name ExternalId='{WM_EXTERNAL_ID}' ) except Exception as e: print(f"Assume role failed: {e}") exit() # Extract temporary credentials from the assumed role response credentials = assumed_role['Credentials'] # 2. Create S3 client using temporary credentials s3 = boto3.client( 's3', aws_access_key_id=credentials['AccessKeyId'], aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'] ) # 3. Upload local file to S3 try: # Path to the local file you want to upload # Replace with your actual file path local_file_path = '{/path/to/your/file}' # Verify if file exists if not os.path.exists(local_file_path): print(f"Error: File '{local_file_path}' does not exist") exit() # Extract filename from path file_name = os.path.basename(local_file_path) # Get current UTC time and format it for directory structure # Format: YYYY/MM/DD utc_now = datetime.now(timezone.utc) time_path = utc_now.strftime("%Y/%m/%d/%h") # Build S3 object key using tenant prefix and time-based path s3_object_key = f"{TENANT_PREFIX}/{time_path}/{file_name}" print(f"Uploading {local_file_path} to s3://{BUCKET_NAME}/{s3_object_key}...") # Perform the file upload to S3 s3.upload_file( Filename=local_file_path, Bucket=BUCKET_NAME, Key=s3_object_key ) print(f"File upload successful: s3://{BUCKET_NAME}/{s3_object_key}") except Exception as e: print(f"File upload failed: {e}")

#!/bin/bash # === Config (edit if needed) === # Client's internal account ID in WorkMagic WM_ACCOUNT_ID="121226" # e.g., 123456 # Client's source bucket (edit this to their actual bucket name) SOURCE_BUCKET_NAME="customer-source-bucket" # === Compute UTC timestamp for the previous hour === DATE_PATH=$(date -u -d "1 hour ago" "+%Y/%m/%d/%H") SOURCE_PATH="s3://${SOURCE_BUCKET_NAME}/${DATE_PATH}/" # Destination bucket and final path DEST_BUCKET="workmagic-inbound-data" DATA_CATEGORY="order-geo-date-data" DEST_PATH="account-121226/{DATA_CATEGORY}/${DATE_PATH}/" # Role info (provided by your platform) ROLE_ARN="arn:aws:iam::588759585297:role/AmazonS3FolderAccess121226" EXTERNAL_ID="ext_id_account_121226" # Replace with default workmagic provided external id # Temp directory TMP_DIR="/tmp/order-upload-121226-${DATE_PATH}/-}" # AWS Region REGION="us-east-1" # === Execution === set -e echo "[1/4] Preparing temp directory: $TMP_DIR" rm -rf "$TMP_DIR" mkdir -p "$TMP_DIR" echo "[2/4] Syncing data from customer bucket: $SOURCE_PATH" aws s3 sync "$SOURCE_PATH" "$TMP_DIR" --region "$REGION" echo "[3/4] Assuming role: $ROLE_ARN" ASSUME_OUTPUT=$(aws sts assume-role --role-arn "$ROLE_ARN" --role-session-name "order-upload-session-121226" --external-id "$EXTERNAL_ID" --region "$REGION") export AWS_ACCESS_KEY_ID=$(echo "$ASSUME_OUTPUT" | jq -r '.Credentials.AccessKeyId') export AWS_SECRET_ACCESS_KEY=$(echo "$ASSUME_OUTPUT" | jq -r '.Credentials.SecretAccessKey') export AWS_SESSION_TOKEN=$(echo "$ASSUME_OUTPUT" | jq -r '.Credentials.SessionToken') echo "[4/4] Uploading to: s3://${DEST_BUCKET}/${DEST_PATH}" aws s3 sync "$TMP_DIR" "s3://${DEST_BUCKET}/${DEST_PATH}" --region "$REGION" echo "✅ Sync complete: ${SOURCE_PATH} → s3://${DEST_BUCKET}/${DEST_PATH}" rm -rf "$TMP_DIR"

from google.cloud import storage import os # Path to the customer's service account key file SERVICE_ACCOUNT_JSON = "order-uploader-key.json" # Your service account key JSON file # Target bucket and destination path DEST_BUCKET = "workmagic-inbound-data" DEST_FOLDER = "account-{WM_ACCOUNT_ID}" DATA_CATEGORY = "{DATA_TYPE}" SUBPATH = "2025/05/27/13" # generate current date time FILENAME = "order_data_0001.json.gz" DEST_OBJECT_PATH = f"{DEST_FOLDER}/{DATA_CATEGORY}/{SUBPATH}/{FILENAME}" # Local file to upload LOCAL_FILE = os.path.join(SUBPATH, FILENAME) def main(): # Authenticate with the customer's service account storage_client = storage.Client.from_service_account_json(SERVICE_ACCOUNT_JSON) # Get the bucket bucket = storage_client.bucket(DEST_BUCKET) # Create a blob and upload the file blob = bucket.blob(DEST_OBJECT_PATH) blob.upload_from_filename(LOCAL_FILE) print(f"✅ Uploaded {LOCAL_FILE} to gs://{DEST_BUCKET}/{DEST_OBJECT_PATH}") if __name__ == "__main__": main()

#!/bin/bash # === Config to edit === WM_ACCOUNT_ID="{WM_ACCOUNT_ID}" SOURCE_BUCKET="customer-source-bucket" # === Compute UTC timestamp for the previous hour === DATE_PATH=$(date -u -d "1 hour ago" "+%Y/%m/%d/%H") # === Construct GCS paths === DEST_BUCKET="workmagic-inbound-data" DEST_FOLDER="account-{WM_ACCOUNT_ID}" DATA_CATEGORY="{DATA_TYPE}" SOURCE_PATH="gs://${SOURCE_BUCKET}/${DATE_PATH}/" DEST_PATH="gs://${DEST_BUCKET}/${DEST_FOLDER}/${DATA_CATEGORY}/${DATE_PATH}/" echo "[INFO] Syncing previous hour's data:" echo " From: ${SOURCE_PATH}" echo " To: ${DEST_PATH}" # === Perform the sync === gsutil -m rsync -r "${SOURCE_PATH}" "${DEST_PATH}"

CREATE OR REPLACE STORAGE INTEGRATION s3_order_export_to_workmagic TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = 'S3' ENABLED = TRUE STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::588759585297:role/AmazonS3FolderAccess121226' STORAGE_ALLOWED_LOCATIONS = ('s3://workmagic-inbound-data/account-121226/') STORAGE_AWS_EXTERNAL_ID = 'ext_id_account_121226';

DESC INTEGRATION s3_order_export_to_workmagic;

DECLARE data_category STRING DEFAULT "order-geo-date-data'; sql STRING; BEGIN LET sql := ' COPY INTO ''s3://workmagic-inbound-data/account-121226/<data_category>/<date_path>/'' FROM ( SELECT OBJECT_CONSTRUCT(*) AS json_row FROM your_schema.your_table WHERE order_update_time > now() - interval 2 hour ) STORAGE_INTEGRATION = s3_order_export_to_workmagic FILE_FORMAT = ( TYPE = ''JSON'' COMPRESSION = ''GZIP'' ) MAX_FILE_SIZE = 1048576 OVERWRITE = TRUE;'; sql := REPLACE(sql, '<date_path>', TO_VARCHAR(CURRENT_TIMESTAMP(), 'YYYY/MM/DD/HH')); EXECUTE IMMEDIATE (sql); RETURN sql; END;

DECLARE run_datetime STRING; DECLARE export_uri STRING; DECLARE export_sql STRING; DECLARE timestamp_str STRING; DECLARE data_category STRING; -- Set data category SET data_category = 'order-geo-date-data'; -- Get current timestamp in format YYYY/MM/DD/HH (UTC) SET date_path = FORMAT_TIMESTAMP('%Y/%m/%d/%H', CURRENT_TIMESTAMP()); -- File timestamp string SET timestamp_str = FORMAT_TIMESTAMP('%Y-%m-%d-%H', CURRENT_TIMESTAMP()); -- Build the export file URI SET export_uri = CONCAT('gs://workmagic-inbound-data-account-121226', data_category, date_path, timestamp_str,'-*.json.gz'); -- Construct the EXPORT DATA statement SET export_sql = ''' EXPORT DATA OPTIONS ( uri = "''' || export_uri || '''", format = "JSON", compression = "GZIP", overwrite = true ) AS ( SELECT * from your_order_data where order_update_time > now() - interval 2 hour ); '''; -- Execute the EXPORT DATA statement EXECUTE IMMEDIATE export_sql;