First time moving from Bash to Python for scripting and looking for feedback/improvements. I've spent the weekend learning about OOP, and am aware that not fully utilizing it in this script — so any feedback is much appreciated!

Summary: Python script runs every 5 minutes, confirms server creation/deletion via OpenStack API functioning correctly.


  1. Check if JSON file exists (this way we can avoid asking for new auth token, as it only expires after 12 hours). If not, get auth token and write to JSON.
  2. Read JSON file, ensure auth token hasn't expired.
  3. Create a new server.
  4. Poll status of server every 5 seconds until 'ACTIVE', or timeout after 2 minutes.
  5. Delete server.

I've redacted some code, but should make sense overall. I intend to add logging after more improvements, but back to watching Python-related videos on YouTube for now.

#!/usr/bin/env python3

from datetime import datetime, timedelta
import json
import logging
import os.path
import sys
import time
import requests

OPENSTACK_USER = 'prometheus'

SERVER_NAME = 'ServerHealthCheck'
SERVER_FLAVOR = 'e4b7e835-927a-4400-9b89-6e93968f0033'
SERVER_IMAGE = '30fb86b3-31d8-4191-be7d-aacf6c317c3e'

SCRIPT_FILE = os.path.basename(__file__)
STATE_FILE = os.path.splitext(SCRIPT_FILE)[0] + '.json'
LOG_FILE = os.path.splitext(SCRIPT_FILE)[0] + '.log'

                    format='%(asctime)s: %(levelname)s: %(message)s')

def _url(path):
    return 'https://example.org' + path

def token_refresh():
    token_request = requests.post(_url(':5000/v3/auth/tokens'), json={
        'auth': {
            'identity': {
                'methods': ['password'],
                'password': {
                    'user': {
                        'name': OPENSTACK_USER,
                        'domain': {'id': 'default'},
                        'password': OPENSTACK_PASSWORD,
            'scope': {
                'project': {
                    'domain': {'id': 'default'},
                    'name': OPENSTACK_PROJECT

    if not token_request.status_code == 201:
        logging.error("Token refresh request returned status code {}\n{}\n"
                      .format(token_request.status_code, token_request.json()))

    token = {'token_content': token_request.headers['X-Subject-Token'],
             'token_expires': token_request.json()["token"]["expires_at"]}

    with open(STATE_FILE, 'w') as json_write:
        json.dump(token, json_write)

def server_create():
    create_request = requests.post(_url(':8774/v2.1/servers'), headers=headers, json={
        'server': {
            'name': SERVER_NAME,
            'flavorRef': SERVER_FLAVOR,
            'imageRef': SERVER_IMAGE

    if not create_request.status_code == 202:
        logging.error("Server create request returned status code {}\n{}\n"
                      .format(create_request.status_code, create_request.json()))

    return create_request.json()["server"]["id"]

def server_status(server):
    status_request = requests.get(_url(':8774/v2.1/servers/' + server), headers=headers)
    return status_request.json()["server"]["status"]

def server_delete(server):
    delete_request = requests.delete(_url(':8774/v2.1/servers/' + server), headers=headers)

    if not delete_request.status_code == 204:
        logging.error("Server delete request returned status code {}\n{}\n"
                      .format(delete_request.status_code, delete_request.json()))

if __name__ == '__main__':
    # Get new auth token if state file does not exist
    if not os.path.isfile(STATE_FILE):

    # Load JSON state file as dictionary
    with open(STATE_FILE, 'r') as f:
        data = json.load(f)

    # Get new auth token if current token expired
    current_gmt = datetime.now() + timedelta(seconds=time.timezone)
    if current_gmt.isoformat() >= data["token_expires"]:

    # Set HTTP headers with auth token
    headers = {'X-Auth-Token': data["token_content"],
               'Content-type': 'application/json'}

    # Create new server
    server = server_create()

    # Poll server status
    status = server_status(server)
    request_seconds = 0
    while not status == 'ACTIVE':
        request_seconds += 5

        if request_seconds >= 120:
            logging.error("Server status {} after 2 minutes".format(status))

        status = server_status(server)

    # Delete server
  • 1
    \$\begingroup\$ Python 2 or 3? Python 3 has async, which can be used instead of time.sleep \$\endgroup\$ – hjpotter92 Nov 14 '18 at 2:44

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.