Streamlining Workspace Operations and More With Python and SingleStore’s Management API

In this blog post, we'll explore a suite of Python functions designed to simplify various tasks like creating, resuming, terminating and resizing workspaces, as well as uploading files to staging areas from S3.

Efficiently managing a database requires powerful tools. When coupled with Python, the SingleStore Management API provides an incredible level of control and automation. In this blog post, we'll explore a suite of Python functions designed to simplify various tasks like creating, resuming, terminating and resizing workspaces, as well as uploading files to staging areas from S3. Before we dive into the functions, ensure you have the following Python libraries imported:

import requests
import json
import re
import boto3
from datetime import datetime

For detailed API documentation, visit the SingleStore Management API Reference

Python-powered workspace management

Create workspaces with a simple function

Kickstart your database operations by setting up a new workspace effortlessly using Python.

def create_workspace(name, size, workspaceGroupID, token):
    url = "<https://api.singlestore.com/v1/workspaces>"
    payload = {
        "enableKai": True,
        "name": name,
        "size": size,
        "workspaceGroupID": workspaceGroupID,
    }
    payload_json = json.dumps(payload)
    headers = {"Content-Type": "application/json", "Authorization":
f"Bearer {token}"}
    response = requests.post(url, headers=headers, data=payload_json)
    return response.text

Tip: to get your workspace group ID, navigate to your workspace group and copy the last part of the URL.

Resuming a workspace

Need to get your suspended workspace back online? Do it in a snap with this function.

def resume_workspace(workspaceID, token):
    url =
f"<https://api.singlestore.com/v1/workspaces/{workspaceID}/resume
>"

    headers = {"Content-Type": "application/json", "Authorization":
f"Bearer {token}"}

    response = requests.post(url, headers=headers)

    return response.text

Tip: to get your workspaceID, click into the ellipsis (Actions) next to your workspace, click on “Edit Workspace Settings” and copy the workspace ID.

Terminating a workspace

Save resources by temporarily terminating inactive workspaces using this straightforward function.

def terminate_workspace(workspaceID, token):
    url =
f"<https://api.singlestore.com/v1/workspaces/{workspaceID}>"

    headers = {"Content-Type": "application/json", "Authorization":
f"Bearer {token}"}

    response = requests.requests("DELETE", url, headers=headers)

    return response.text

Resizing a workspace

Adjust the size of your workspace easily to match your current needs, ensuring efficient resource utilization and performance optimization.

def resize_workspace(workspaceID, size, token):
    url =
f"<https://api.singlestore.com/v1/workspaces/{workspaceID}>"

    headers = {"Content-Type": "application/json", "Authorization":
f"Bearer {token}"}

    payload = {
      "size": size
    }

    response = requests.patch(url, headers=headers,
data=json.dumps(payload))

    return response.text

Tip: visit our sizing page for a list of supported sizes (i.e., S-00, S-1, S4, etc.)

Streamline file uploads to Stage

SingleStore Stage allows you to load files in a staging area attached to your workspace group, helping you quickly get your data into SingleStore (read more about it here).. You can either upload a local file, or upload a file in a S3 bucket.

Local file upload:

def upload_to_staging(workspaceGroupID, stagePath, fileName,
filePath, token):
    url =
f"<https://api.singlestore.com/v1/stages/{workspaceGroupID}/fs/{s
tagePath}>"

    headers = {"Authorization": f"Bearer {token}"}

    # If do not want to specify fileName, only filePath
    # fileName = os.path.basename(filePath)

    files = [("file", (fileName, open(filePath, "rb")))]

    response = requests.request("PUT", url, headers=headers,
files=files)

    return response.text

S3 file upload:

def upload_to_staging(workspaceGroupID, bucketName, filepath,
stagePath, awsKeyID, awsSecret, awsToken, token):
    url =
f"<https://api.singlestore.com/v1/stages/{workspaceGroupID}/fs/{s
tagePath}>"

    headers = {"Authorization": f"Bearer {token}"}

    # Accessing the S3 buckets using boto3 client
    s3 = boto3.client('s3',
                        aws_access_key_id=awsKeyID,
                        aws_secret_access_key=awsSecret,
                        aws_session_token=awsToken)

    # Download file from S3
    s3.download_file(bucketName, filepath, 'localfile')

    # Open the file in binary mode
    with open('localfile', 'rb') as f:
        files = {'file': f}
        response = requests.put(url, headers=headers, files=files)

    return response.text

Easy access to billing information

Staying on top of your billing information is crucial. With our Python function, accessing your billing details is just a function call away.

def get_billing(startTime=None, endTime=None, aggregateBy="day"): #
can change aggregateBy to "hour" or "month"
    # format time in ISO 8601 format
    def format_time(time):
        iso8601_pattern =
re.compile(r"^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z$")
        if iso8601_pattern.match(time):
            formatted_time = format_time
        else:
            try:
                parsed_time = datetime.strptime(time, "%Y-%m-%dT%H:%M:%SZ")
                formatted_time = parsed_time.strftime("%Y-%m-%dT%H:%M:%SZ")
            except ValueError:
                try:
                    # if no time provided, add 'T00:00:00Z'
                    parsed_date = datetime.strptime(time, "%Y-%m-%d")
                    formatted_time =
parsed_date.strftime("%Y-%m-%dT00:00:00Z")
                except Exception as e:
                    raise ValueError(
                        "Invalid date time format. Please provide a valid ISO 8601
formatted time."
                    )
        return formatted_time

    formatted_startTime = format_time(startTime)
    formatted_endTime = format_time(endTime)

    url =
f"<https://api.singlestore.com/v1/billing/usage?startTime={format
ted_startTime}&endTime={formatted_endTime}&aggregateBy={aggregate
By}>"
    headers = {"Content-Type": "application/json", "Authorization":
f"Bearer {token}"}

    response = requests.get(url, headers=headers)
    return response.text

This function will return a JSON object detailing the compute credits and storage usage for each workspace in the specified workspace group. You can then plot graphs to visualize — for instance, the compute credits consumed over time by each workspace. The following code snippet will aggregate this consumption by day:

# Plotting the Compute Credits Consumed Per Day for Each Workspace

import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime

json_data = get_billing(startTime="2023-10-25",
endTime="2023-10-28")
data = json.loads(json_data)

# Function to parse the date-time format in the data
def parse_datetime(dt_str):
    return datetime.strptime(dt_str, '%Y-%m-%d %H:%M:%S %z UTC')

# Complete data extraction
all_usage = []
for billing in data['billingUsage']:
    all_usage.extend(billing['Usage'])

# Convert to DataFrame
df_all = pd.DataFrame(all_usage)

# Convert endTime, startTime, and value to appropriate types
df_all['endTime'] = df_all['endTime'].apply(parse_datetime)
df_all['value'] = pd.to_numeric(df_all['value'])

# Extracting all distinct workspace names
all_resource_names = df_all['resourceName'].unique()

# Creating a bar plot for each distinct workspace name
plt.figure(figsize=(15, 10))

# Offset to position bars next to each other
bar_width = 0.04  # Adjusting bar width for potentially large number of
resources

# Plotting individual bars for each workspace
for idx, resource_name in enumerate(all_resource_names):
    # Filter data for each workspace
    filtered_data = df_all[df_all['resourceName'] == resource_name]

    # Group by date and sum the values
    grouped_data = filtered_data.groupby('endTime').agg({'value':
'sum'}).reset_index()
    # Plotting
    plt.bar(grouped_data['endTime'] + pd.to_timedelta(idx *
bar_width, unit='D'),
            grouped_data['value'],
            width=bar_width,
            label=resource_name)

# Adjust x-axis to accommodate the bar width
plt.xticks(rotation=45)
plt.title('Compute Credits Consumed Per Day for Each Workspace')
plt.xlabel('Date')
plt.ylabel('Compute Credits')
plt.legend(title='Resource Name', bbox_to_anchor=(1.05, 1),
loc='upper left')
plt.tight_layout()
plt.show()

You will get a plot that looks something like this:

Why use these Python functions?

These Python functions serve as practical examples of how to interact with the SingleStore Management API. They are designed to be:

• User friendly. Even if you're not familiar with the API, you can start using these functions right away.
• Timesaving. Automate repetitive tasks and focus on more important aspects of your work.
• Customizable. Feel free to modify these functions to better fit your specific requirements.

Learn more

For a comprehensive understanding of what the SingleStore Management API offers, including detailed endpoint descriptions, check out the API Overview. You can put this all to the test by activating your SingleStore trial — including with our newly released Free Shared Tier.

The SingleStore Management API offers robust capabilities for managing your database environment. By leveraging these Python functions, you can automate and streamline your workspace operations and file management tasks. Whether you're adjusting workspace sizes, managing workspaces or handling file uploads, these tools are designed to enhance your productivity and simplify your database management journey.