·舟山詩詞·淘海洗玉集 – My Poems, and etc.

Python Class in Design

with 2 comments

Abstract: Review the discussion of comparing designs between Python class and module interfaces, and provide a more complicate demo to illustrate how to use Python classes to resolve dependencies, in following sections.

In last blog (“Python class vs module”), the discussion has been slightly in favor of using Python class mechanism to inject dependencies (e.g. settings for swift store, in the demo) via a natural OOP approach, instead of traditional python (scripting) module methods (either passing dependencies in parameters or more complicatedly introducing SwiftConfig class and @checks_config decorator in swift module). The idea of using class is to have some abstraction in design, so that an application can program to interfaces (e.g. ISettings and IStore) other than concrete implementations.

Let’s say to program the interface of IStore. The contracts are listed here –

  save_file(file_name, file_contents)

By one implementation, of using swiftclient on object store, the Swift class has encapsulated dependencies in constructor and merely exposed/implemented methods per above IStore contracts. In other situation or project, if backend store happened to be a database (or some file system), the implementation could be easily swapped by a concrete DBStore or FileStore class, certainly with different signatured constructor (since dependencies vary), but remained same interfaces so that application needs less code and logic change, and less regression.

In this context, settings (e.g. user name, url, auth token, container name for swift, database connection string for db store, or a base directory for file system) are more about how a concrete implmentation exactly set up before the interface can be called. Such dependencies vary from implementations, and should be separated from major business logic (which should care only about IStore interface). Without encapsulation, the interface method signature would have to change for different store. For example, to get a file, a container name is required for swift store, but a base directory is needed for file store.

1. Class Design

Continued on the topic, the next demo, also in python code, is to deploy package to Helion Development Platform (by using Application Lifecycle Service, a.k.a. ALS – a Cloud Foundry-based, managed runtime environment). The procedure of the deployment is actually processing a batch of Helion CLI (cf equivalent tool) commands running against an ALS endpoint. The whole process could have any numbers of sequential steps (each runs a command, with arguments, and returns exit code). If any command failed, the process would stop and fail; otherwise, the process continued in success state to the last command.

During the process (by an artificial deployment identifier), it would be ideal to log the status of each step with certain details. And the whole execution should be running in a fork thread or process in a typical GUI or web application so that the main process (likely the UI) won’t be frozen and held on waiting for the execution completes. The status (with deployment info, might include some history and package details) is recorded/saved to a store (e.g. a swift or database) so that the main UI can use another thread or the web can call a (RESTful) service to check on the progress asynchronously per a deployment identifier.

Although not quite enough “acceptance criteria” for a real project, the above requirements have provided some basic to start a design: a Deployment can launch a Process and also call on an IStatus (which is separated from how status is recorded and retrieved). There is an encapsulated Package model could be part of Deployment and passed along the process. The Process is the interface to execute the Batch. And in this case, since ALS is a concrete service that the deployment process will target to, it can be bound to the Process (but not to Deployment or Batch). The design is fully described as in below.

1.1. Batch

Usage: A class module contains a batch (a series of shell commands, with CWD, target status on success, accepted exit code, or allowing non-zero exit code)
Dependencies: CWD (default working directory specified)

  add(self, command, expected_status, description='', accept_error=False)
1.2. BatchProcess (optionally implements an IProcess)

Usage: A class module represent a batch process (e.g. loop through each command in a batch sequentially and use subprocess to execute the command)
Dependencies: Batch instance, and IStatus interface (or a function pointer to set status record on each step of the process)

1.3. HelionCliComposer

Usage: A class module wrapper to compose Helion CLI command
Dependencies: ALS cluster endpoint, user name (admin email), login password, and optional CWD (working directory)

1.4. Package (Model)

Usage: A model class represent package-related information (e.g. manifest, package/file type, destination, etc.)
Dependencies: None, or a more detailed package manifest (including type and destination)

1.5. DeploymentStatus (IStatus interface)

Usage: A class module represent an interface of getting and setting status (against, e.g. swift or db store)
Dependencies: Package, IStore (Swift or DB store for status record)

  get_status(self, deployment_id)
  set_status(self, deployment_id, status)
1.6. Deployment

Usage: A class module represent a deployment process (e.g. process package and deployment info, build batch commands, and kick off batch process)
Dependencies: Package, HelionCliComposer, IProcess, and IStatus


2. Commonly-used Functions

Based on the Swift class (in last blog), it is easy and straightforward to add more functions related to the store. Assuming to use the store saving both packages and deployment state records, the following piece is a partial of Swift class to include 3 more methods: check_file_exists, check_package_exists and get_deployment_list. Noticing latter two methods have some business logic (about package and deployment) that may not belong to a “pure” IStore interface, it would be a design decision how services are structured and if they should be in Deployment or another middle tier class.

See (partial of Swift class) –

    def check_file_exists(self, file_name):
        if (not self.check_container()):
            return False
        result = self.connection.get_container(
                container_name, full_listing=True)
        for file in result[1]:
            if (file['name'] == file_name):
                return True
        return False

    def check_package_exists(self, package_name):
        file_name = '{0}.tar.gz'.format(package_name)
        return self.check_file_exists(file_name)

    def get_deployment_list(self):
        deployment_list = []
        result = self.connection.get_files_in_container()
        regex = re.compile('^deployment_(.+).json$')
        for file in result:
            filename = file['name']
            re_match =
            add_flag = \
                file['content_type'] == 'application/json' and \
                re_match is not None
            if (add_flag):
                    file_contents = self.get_file_contents(filename)
                    item = json.loads(file_contents)
                except Exception:
        return deployment_list

Another helper module is, which could have commonly used functions that do not belong to any of classes in this demo. There is no need to wrap these functions into a class. In other OOP language (like Java or C#), they are usually grouped as public static methods. Python module serves the same perfectly here. The module also includes a get_store method. This is to demonstrate as a factory to construct a IStore object, especially in a multi-project environment when IStore implementations come from a common namespace but dependencies (e.g. settings) in application domain.


import re
import StringIO
import shutil
import tarfile

from config import settngs
from keystone import get_auth_token
from swift_class import get_swift, Swift
from logging import getLogger
logger = getLogger(__name__)

def delete_directory_tree(dir_path):
    Cleanup a directory
    if (dir_path):
  'Deleting {0}'.format(dir_path))
            # Clean up working directory
            shutil.rmtree(dir_path, ignore_errors=True)
  'Deleted dir: {0}'.format(dir_path))
        except Exception as e:
            err_message = \
                'Failed to clean up working directory "{0}".' \

def extract_manifest_from_package(file_contents):
    Extract the manifest from the vendor package
    manifest_regex = '^.+[/]manifest.json$'
    pattern = re.compile(manifest_regex, re.IGNORECASE)

    # tarfile -
    manifest = None
            fileobj=StringIO.StringIO(file_contents)) as tar_package:
        for tarinfo in tar_package.getmembers():
            if (
                manifest = tar_package.extractfile(
    return manifest

def get_store():
    Get a Swift instance per application settings
    auth_token = get_auth_token()
    container = settings('swift_container')
    swift_url = settings('swift_url')
    swift = Swift(auth_token, swift_url, container)
    return swift

The last piece in this discussion section is Batch and BatchProcess. Both of them are very self-contained and have nothing specifically related to major business logic (Deployment in this case). The separation here is used to isolate each problem domain without too much dependencies at interface level. Envision that the deployment business might need to target on a different platform or require to call a RESTful service instead of a batch of commands, the deploy interface in Deployment would be rewritten to call a different process. The deploy call could have minimum, or even no code change (if an IProcess is defined).

See (Batch and BatchProcess) –

import json
import os
import subprocess
import threading

from logging import getLogger
logger = getLogger(__name__)

class Batch(object):
    def __init__(self, cwd):
        Initialize an instance of Batch
            cwd: current working directory (where the batch to be executed)
        self.batch_cmds = []
        self.cwd = os.path.abspath(os.path.expanduser(cwd))

    def add(self, status, command, accept_error=False):
        Add a command to batch, with expected status on success, and
        optionally allowing non-zero exit code by accept_error=True
            'accept_error': accept_error,
            'command': command,
            'cwd': self.cwd,
            'exit_code': 0,
            'status': status,
            'stdout': '',

    def clear(self):
        self.batch_cmds = []

class BatchProcess(object):
    def __init__(self, batch, set_status_func):
        Initialize an instance of BatchProcess
        self.batch_cmds = batch.batch_cmds
        self.set_status = set_status_func
        self.started = False
        self.success = False

    def execute(self):
        Start to execute a batch process
        can_continue = True
        self.started = True
        for next_cmd in self.batch_cmds:
  'next cmd:\n{0}'.format(
                json.dumps(next_cmd, indent=2, sort_keys=True)))
            accept_error = next_cmd['accept_error']
            cmd = next_cmd['command']
            # ToDo [zhuyux]: add timeout mechnisam
            proc = subprocess.Popen(
            next_cmd['stdout'] = proc.communicate()[0]
            stdout = next_cmd['stdout'].decode('string_escape')
            exit_code = proc.returncode

            if (accept_error or exit_code == 0):
                logger.error('Exit code {0} from {1}'.format(exit_code, cmd))
                next_cmd['exit_code'] = exit_code
                can_continue = False

        self.set_status('SUCCESS' if can_continue else 'FAILED')
        self.success = can_continue
        return can_continue

3. Source Code

This section mainly lists rest of the source code at core business of the Deployment. By this far, it should be clear to see how a class is designed to be highly cohesive (to its own problem domain) but also loosely decoupled from other classes, modules, or layers. Dependencies between each class/module are kept at minimum by object constructor or a factory, while interfaces are maintained clean and consistent regardless of concrete implementations. Services are self-contained and swappable without affecting too much on other part of the application. The design thought is for Python classes, but applies as generic in any programming practice.

See (HelionCliComposer class) –
import os

class HelionCliComposer(object):
    def __init__(self, endpoint, username, password, cwd=None):
        Initialize an instance of HelionCliComposer
        self.cwd = None
        if (cwd is not None):
            self.cwd = os.path.abspath(os.path.expanduser(cwd))
        self.endpoint = endpoint
        self.username = username
        self.password = password

    def get_delete_cmd(self, name):
        return [
            'helion', 'delete',
            '--target', '{0}'.format(self.endpoint),
            '-n', '{0}'.format(name)]

    def get_list_cmd(self):
        return ['helion', 'list', '--target', '{0}'.format(self.endpoint)]

    def get_login_cmd(self):
        return [
            'helion', 'login',
            '--credentials', 'username: {0}'.format(self.username),
            '--password', '{0}'.format(self.password),
            '--target', '{0}'.format(self.endpoint)]

    def get_logout_cmd(self):
        return ['helion', 'logout']

    def get_push_cmd(self, name, path):
        if (self.cwd is not None):
            path = '{0}/{1}'.format(self.cwd, path)
        return [
            'helion', 'push',
            '--target', '{0}'.format(self.endpoint),
            '--as', '{0}'.format(name),
            '--path', '{0}'.format(path),

    def get_target_cmd(self):
        return ['helion', 'target', self.endpoint]

See (Package class) –

import re
import os

from logging import getLogger
logger = getLogger(__name__)

class Package(object):
    def __init__(self, package_id, package_path, endpoint_url=None):
        Initialize an instance of Package
            package_id: package id or name
            package_path: full path of the package (including file name)
        """ = package_id
        self.file_name = os.path.basename(package_path) = self.get_package_name(package_path)
        self.path = os.path.abspath(os.path.expanduser(package_path))
        self.destination = self.get_destination(endpoint_url,
        self.cwd = os.path.dirname(self.path)

    def get_destination(self, endpoint_url, package_name):
        Get package destination url from endpoint and package name
        dest = ''
        if (endpoint_url):
            regex = re.compile('^(http[s]?://)api\.(.+)$')
            re_match ='/'))
            if (re_match is not None):
                prot =
                addr =
                dest = '{0}{1}.{2}'.format(prot, package_name, addr)
        # returning package destination url
        return dest

    def get_package_manifest_filename(self):
        Get package manifest filename (e.g. foo.json) without path
        return '{0}.json'.format(

    def get_package_name(self, package_path):
        Get package name (e.g. foo) from package path (e.g. '/path/foo.tar.gz')
        pkg_file = os.path.basename(package_path)
        pkg_name = os.path.splitext(os.path.splitext(pkg_file)[0])[0]
        return pkg_name

See (DeploymentStatus class) –

import json
import os

from datetime import datetime
from time import gmtime, strftime

from logging import getLogger
logger = getLogger(__name__)

class DeploymentStatus(object):
    def __init__(self, package=None, store):
        Initialize an instance of DeploymentStatus
        self.package = package
        self.destination = '' if package is None else package.destination
        self.package_name = 'N/A' if package is None else = store

    def get_all(self):

    def get_deployment_filename(self, id):
        filename = 'deployment_{0}.json'.format(id)
        return filename

    def get_status(self, id):
        get status record (as json object) by deployment id
        result = {
            'deploy_id': id,
            'deploy_status': '',
            'datetime': '',
            'destination': self.destination,
            'history': [],
            'package': self.package_name}
            filename = self.get_deployment_filename(id)
            contents =

            if (contents):
                # logger.debug('Deployment status: {0}'.format(contents))
                result = json.loads(contents)
        except Exception as e:
            logger.exception("Failed to get status for {0}.\n".format(id))
        # logger.debug('Deployment result: {0}'.format(result))
        return result

    def set_status(self, id, status):
        set status record (json file) by deployment id and status (string)
        """'======= Setting status: "{0}" =======\n'.format(status))

        result = self.get_status(id)

        date_time = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")
        history = result['history']
        record = '{0} ~ {1}'.format(date_time, status)
        if (type(history) is list):
        else:  # creating a history list
            history = [record]

        result['deploy_id'] = id
        result['deploy_status'] = status
        result['datetime'] = date_time
        result['destination'] = self.destination
        result['package'] = self.package_name
        result['history'] = history

        filename = self.get_deployment_filename(id)
        contents = json.dumps(result, sort_keys=True), filename, contents)

        return result

See (Deployment class) –

import os
import shutil
import tempfile

from multiprocessing import Lock, Process, Queue
from batch import Batch, BatchProcess

from utils import delete_directory_tree
from logging import getLogger
logger = getLogger(__name__)

class Deployment(object):
    def __init__(
            package, cli_composer, deploy_status,
        Initialize an instance of Deployment
        import uuid
        if (use_package_path):
            self.batch = Batch(package.cwd)
            self.batch = Batch(tempfile.mkdtemp())
        self.cli_composer = cli_composer
        self.cwd = self.batch.cwd
        self.cwd_use_package_path = use_package_path
        self.deployed = False
        self.deployment_id = '{0}'.format(uuid.uuid1())
        self.deploy_status = deploy_status
        self.package = package =  # backend store
        self.started = False

    def cleanup(self):
        Cleanup Deployment BatchPrcess working directory
  'Deleting deployment cwd={0}'.format(self.cwd))
            # Clean up working directory
  'Deleted deploy deployment cwd.')
        except Exception as e:
            err_message = \
                'Failed to clean up deployment cwd "{0}".' \

    def deploy(self):
        Start a Deployment process
        if (self.started):
            err = 'Deployment {0} already started'.format(self.deployment_id)
            raise Exception(err)


            self.started = True
            self.download_package() # preparing package

  'Starting deployment ...')
            process = BatchProcess(self.batch, self.set_status)
            logger.debug('Batch process: {0}'.format(process))
            self.deployed = process.execute()
        except Exception as e:
            err_message = 'Exception on BatchProcess execution.'
  'DONE deployment - {0}'.format(self.deployment_id))

    def download_package(self):
        if (self.cwd_use_package_path):
            pkg_filename = self.package.file_name
            pkg_contents = store.get_file_contents(pkg_filename)
  'Downloading package {0} to {1}...'.format(
                pkg_filename, self.package.path))
            with open(self.package.path, 'w') as package_file:
                # write the package as a tar.gz into deployment cwd
        return self.package.path

    def get_deployment_batch(self):
        Get a batch of commands for the deployment
        pkg_path = self.package.path
        pkg_name =

        # add unpacking script to batch'Adding batch to unpack {0} from {1}'.format(
            pkg_name, pkg_path))

        # add deployment script to batch
        self.batch.add('TARGET', self.cli_composer.get_target_cmd())
        self.batch.add('LOGIN', self.cli_composer.get_login_cmd())
            'REMOVED', self.cli_composer.get_delete_cmd(pkg_name), True)
        self.batch.add('LIST', self.cli_composer.get_list_cmd())
        self.batch.add('DEPLOYED', self.cli_composer.get_push_cmd(
            pkg_name, '{0}'.format(pkg_name)))
        self.batch.add('NEWLIST', self.cli_composer.get_list_cmd())
        self.batch.add('DIR', ['ls', '-al'])

    def get_package_batch(self):
        Get a batch of commands for preparing the package
        dst_path = self.cwd
        src_path = self.package.path
        pkg_name =

        # no need this copy command if package path is used as cwd
        if (not self.cwd_use_package_path):
            copy_cmd = [
                'cp', '-rf',
            self.batch.add('COPY', copy_cmd)

        view_cmd = [
            'tar', '-tvf',
        # Assume any foo.tar.gz contains -
        #   - foo/foo.tar.gz (the package to deploy)
        #   - manifest.json
        unpack_cmd = [
            'tar', '-zxvf',
        xtract_cmd = [
            'tar', '-zxvf',
            '{0}/{1}/{2}.tar.gz'.format(dst_path, pkg_name, pkg_name)]
        dir_cmd = [
            'ls', '-al',
            '{0}/{1}'.format(dst_path, pkg_name)]
        self.batch.add('PREVIEW', view_cmd)
        self.batch.add('UNPACK', unpack_cmd)
        self.batch.add('EXTRACT', xtract_cmd)
        self.batch.add('DIR', dir_cmd)

    def get_status(self):
        '''get status by self.deployment_id
        return self.deploy_status.get_status(self.deployment_id)
        # return status

    def set_status(self, status):
        '''set status by self.deployment_id
        self.deploy_status.set_status(self.deployment_id, status)

See (app module) –

import json
import shutil
import tempfile
import traceback

from multiprocessing import Process
from subprocess import call, check_output, CalledProcessError

from deploy import Deployment
from deploy_status import DeploymentStatus
from helion_cli import HelionCliComposer
from package import Package
from utils import delete_directory_tree, get_store
from logging import getLogger
logger = getLogger(__name__)

def deploy_package(package_name, endpoint_url, username, password):
    Deploy a package into destination (e.g. ALS/Cloud Foundry)
        package_name - the name of the package to deploy
        endpoint_url - the destination (e.g. ALS/Cloud Foundry) endpoint URL
                       ie: ''
        username - the user name (admin email) for destination login
        password - the password for destination login
    store = get_store()

    if (not store.check_package_exists(package_name)):
        return {'status': 404}

    cwd = ''
        cwd = tempfile.mkdtemp()
        pkg_filename = '{0}.tar.gz'.format(package_name)
        package_path = '{0}/{1}'.format(cwd, pkg_filename)
        package = Package(package_name, package_path, endpoint_url)

        # instantiate a cli composer
        composer = HelionCliComposer(endpoint_url, username, password)

        deploy_status = DeploymentStatus(package, store)
        deployment = Deployment(package, composer, deploy_status, True)
        deployment_id = deployment.deployment_id


        # Start a new process to execute the deployment
        process = Process(
        process.start()'Deployment {0} started for {1}.'.format(
            deployment_id, package_name))

        return {
            'status': 201,
            'deployment_id': deployment_id,
            'package': package_name}

    except Exception as e:
        stack_info = traceback.format_exc()
        error_message = "Exception on deploy {0}. Details:\n{1}".format(
            package_name, stack_info)
        return {'status': 500, 'errors': error_message}

def get_status(id):
    Get the deployment status by id
    try:"======= deployment::get_status =======")
        store = get_store()
        deploy_status = DeploymentStatus(store=store)
        result = deploy_status.get_status(id)

        logger.debug('Deployment result: {0}'.format(result))
        if result == {} or not result['deploy_status']:
            return {'status': 404}
            return {'status': 200, 'data': result}
    except Exception as e:
        stack_info = traceback.format_exc()
        error = "Exception on getting deployment status"
        error_message = "{0} for {1}. Details:\n{2}".format(
            error, id, stack_info)
        return {'status': 500, 'errors': error_message}

2 Responses

Subscribe to comments with RSS.

  1. Reblogged this on Dinesh Ram Kali..


    2015-08-02 at 22:54

  2. […] Next topic (“Python Class in Design“) of this “Let Code Speak” series will use a more complicate task to demonstrate how python classes are used in OOP design. All source code in this demo are downloadable at here. […]

Leave a Reply

Fill in your details below or click an icon to log in: Logo

You are commenting using your account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s

%d bloggers like this: