Trigger storage snapshot via Python

This is a net script I have dug for one of our user asking on support how to easily trigger snapshot for multiple tables. Please not this could be (as python transformation) orchestrated for instance in your weekly/monthly flows to create snapshots of critical data that will last longer than automatic time-travel functionality.

Please note it is not ideal, since it exposes the token (but you can either create a dedicated one, or reset your personal after use).

All you have to use this in a python transformation and replace is the token and the table list:

table_names = ["out.c-bitbucket.repositories", "out.c-bitbucket.repository_stats"]

The whole script is following:

import requests
import sys
 
# Please enter token here:
TOKEN = "{PUT TOKEN HERE}"
 
# List table names here: ["table1","table2","etc."]
table_names = ["out.c-bitbucket.repositories", "out.c-bitbucket.repository_stats"]
 
def send_request(table_name):
    # Create Table Snapshot
    # POST https://connection.keboola.com/v2/storage/tables/out.c-data_model.carpark/snapshots
    call_url = ("https://connection.keboola.com/v2/storage/tables/"+str(table_name)+"/snapshots")
    try:
        response = requests.post(
            url = call_url,
            headers = {
                "X-StorageApi-Token": TOKEN,
                "Content-Type": "application/x-www-form-urlencoded",
            },
            data={"description": "orchestration",},
        )
        print('Response HTTP Status Code: {status_code}'.format(
            status_code=response.status_code))
        print('Response HTTP Response Body: {content}'.format(
            content=response.content))
    except requests.exceptions.RequestException:
        print('HTTP Request failed')
        sys.exit(1)
 
 
for table in table_names:
      send_request(table)

If you have project on other maintainer, such as PAYG in N-EU stack (Azure), please use corresponding base URL.

Possible improvements:

Create a dedicated component to protect the token (any volunteers?)
Combine this with storage API to get the list of all tables so you can just specify the bucket and call the day...

Hope this helps!