-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added a Python example to launch DB, test DB connection and terminate…
…. Including a colab notebook.
- Loading branch information
Showing
2 changed files
with
204 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,203 @@ | ||
# Launch DB using Python | ||
|
||
[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/drive/1j8sdpIZQ0J-E4fvGuhaXJTsVN6s5ZecW?usp=sharing) --> Easy Launch on Google Colab | ||
<!-- TODO: Publish notebook to a public SkySQL git repo and change above URL --> | ||
|
||
The Python example below shows how to launch a DB service and connect to it using the SkySQL DBaaS API. | ||
|
||
You need to configure the following environment variables: | ||
* `API_KEY` | ||
* `SKYSQL_CLIENT_IP` | ||
|
||
### **Step 1: Generate API Key** | ||
|
||
1. Go to [SkySQL API Key management page](https://app.skysql.com/user-profile/api-keys) and generate an API key. | ||
|
||
2. Export the value from the token field to an environment variable API_KEY: | ||
|
||
```bash | ||
export API_KEY='... key data ...' | ||
``` | ||
|
||
The `API_KEY` environment variable will be used in the subsequent steps. | ||
|
||
Check any existing services: | ||
```bash | ||
curl --request GET 'https://api.skysql.com/provisioning/v1/services' \ | ||
--header "X-API-Key: $API_KEY" | ||
``` | ||
|
||
### **Step 2: Determine the Client IP Address** | ||
|
||
When your new service is created, your client can only connect through the service's firewall if the client IP address is in the service's IP allowlist. | ||
|
||
Before creating the new service, determine the public IP address of your client host and save it to the `SKYSQL_CLIENT_IP` environment variable. | ||
|
||
If you are not sure of your public IP address, you can use a lookup service, such as `checkip.amazonaws.com`: | ||
|
||
```bash | ||
export SKYSQL_CLIENT_IP=`curl -sS checkip.amazonaws.com` | ||
``` | ||
|
||
### **Step 3: Create the Service using Python** | ||
|
||
The following Python script has been tested with Python 3.11 | ||
|
||
```python | ||
import os | ||
import time | ||
import requests | ||
import json | ||
import sys | ||
import argparse | ||
import mysql.connector | ||
from mysql.connector import Error | ||
# Set your API key and client IP | ||
API_KEY = os.getenv('API_KEY') | ||
CLIENT_IP = os.getenv('SKYSQL_CLIENT_IP') | ||
## Name of the service to be created | ||
SKYDB_SERVICE_NAME = 'my-first-skysql-db' | ||
# Headers for the API requests | ||
headers = { | ||
'Content-Type': 'application/json', | ||
'X-API-Key': API_KEY | ||
} | ||
# Function to launch a SkySQL DB service | ||
def launch_skysql_db(): | ||
url = "https://api.skysql.com/provisioning/v1/services" | ||
payload = { | ||
"service_type": "transactional", | ||
"topology": "es-single", | ||
"provider": "gcp", | ||
"region": "us-central1", | ||
"architecture": "amd64", | ||
"size": "sky-2x8", | ||
"storage": 100, | ||
"nodes": 1, | ||
"name": f"{SKYDB_SERVICE_NAME}", | ||
"ssl_enabled": True, | ||
"allow_list": [ | ||
{ | ||
"comment": "Describe the IP address", | ||
"ip": f"{CLIENT_IP}/32" | ||
} | ||
] | ||
} | ||
try: | ||
response = requests.post(url, headers=headers, data=json.dumps(payload)) | ||
response.raise_for_status() | ||
return response | ||
except requests.exceptions.RequestException as e: | ||
print(f"Failed to launch SkySQL DB service: {e}") | ||
return None | ||
# Function to monitor the status of the service | ||
def monitor_status(service_id): | ||
url = f"https://api.skysql.com/provisioning/v1/services/{service_id}" | ||
while True: | ||
response = requests.get(url, headers=headers) | ||
status = response.json().get('status') | ||
if status == 'ready': | ||
print("Service is ready.. ") | ||
print(response.json()) | ||
return response.json() | ||
print("Service launch started .. will wait for 30 secs and check again...") | ||
time.sleep(30) | ||
# Function to test the connection to the service | ||
def test_connection(connection_properties): | ||
try: | ||
connection = mysql.connector.connect( | ||
host=connection_properties['host'], | ||
user=connection_properties['user'], | ||
password=connection_properties['password'], | ||
) | ||
if connection.is_connected(): | ||
print("Connection successful") | ||
connection.close() | ||
except Error as e: | ||
print(f"Error: {e}") | ||
# Function to fetch security credentials for the service | ||
def fetch_credentials(service_id): | ||
url = f"https://api.skysql.com/provisioning/v1/services/{service_id}/security/credentials" | ||
try: | ||
response = requests.get(url, headers=headers) | ||
response.raise_for_status() | ||
return response.json() | ||
except requests.exceptions.RequestException as e: | ||
print(f"Failed to fetch credentials: {e}") | ||
return None | ||
# Main function to execute the steps | ||
def main(): | ||
# Step 0: Parse command-line arguments | ||
parser = argparse.ArgumentParser(description="Launch SkySQL DB or use an existing service") | ||
parser.add_argument('--service_id', type=str, help="Optional service ID to skip launching a new service") | ||
args = parser.parse_args() | ||
# Step 1: Either use the provided service_id or launch a new SkySQL DB | ||
if args.service_id: | ||
print(f"Using provided service ID: {args.service_id}") | ||
service_id = args.service_id | ||
else: | ||
service_response = launch_skysql_db() | ||
# Check if the response status code is within the 2xx range (success) | ||
if not (200 <= service_response.status_code < 300): | ||
print("Failed to launch SkySQL DB service") | ||
print(service_response.json()) # Print detailed error message | ||
return | ||
service_json = service_response.json() | ||
service_id = service_json.get('id') | ||
if not service_id: | ||
print("Service launched, but could not retrieve service ID") | ||
return | ||
# Step 2: Monitor the status until 'ready' | ||
service_details = monitor_status(service_id) | ||
if not service_details: | ||
print("Service did not become ready") | ||
return | ||
# Step 3: Obtain the connection properties (FQDN and port) | ||
fqdn = service_details.get('fqdn') | ||
ports = service_details['endpoints'][0]['ports'] | ||
# Find the port where name is 'readwrite' | ||
readwrite_port = next((port['port'] for port in ports if port['name'] == 'readwrite'), None) | ||
if not fqdn or not readwrite_port: | ||
print("Failed to retrieve FQDN or port information.") | ||
return | ||
# Step 4: Obtain the default credentials (username and password) | ||
credentials = fetch_credentials(service_id) | ||
if not credentials: | ||
print("Failed to retrieve credentials") | ||
return | ||
connection_properties = { | ||
'host': fqdn, | ||
'port': readwrite_port, | ||
'user': credentials.get('username'), | ||
'password': credentials.get('password'), | ||
} | ||
# Step 5: Test the connection to the DB | ||
test_connection(connection_properties) | ||
if __name__ == "__main__": | ||
main() | ||
``` | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters