boat+hill

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

Panama Cruise 2015

leave a comment »

Written by Boathill

2015-04-02 at 23:00

Python Class in Design

with one comment


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 –

  get_files_list()
  get_file_contents(file_name)
  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)
Interfaces:

  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)
Interfaces:

  execute(self)
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)
Interfaces:

  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
Interfaces:

  deploy(self)

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 swift.py (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 = regex.search(filename)
            add_flag = \
                file['content_type'] == 'application/json' and \
                re_match is not None
            if (add_flag):
                try:
                    file_contents = self.get_file_contents(filename)
                    item = json.loads(file_contents)
                    deployment_list.append(item)
                except Exception:
                    continue
        return deployment_list

Another helper module is utils.py, 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 utils.py 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.

See utils.py

# utils.py
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):
        try:
            logger.info('Deleting {0}'.format(dir_path))
            # Clean up working directory
            shutil.rmtree(dir_path, ignore_errors=True)
            logger.info('Deleted dir: {0}'.format(dir_path))
        except Exception as e:
            err_message = \
                'Failed to clean up working directory "{0}".' \
                .format(dir_path)
            logger.exception(err_message)


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 - https://docs.python.org/2/library/tarfile.html
    manifest = None
    with tarfile.TarFile.open(
            mode='r:gz',
            fileobj=StringIO.StringIO(file_contents)) as tar_package:
        for tarinfo in tar_package.getmembers():
            if (pattern.search(tarinfo.name)):
                manifest = tar_package.extractfile(tarinfo.name).read()
                break
    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.py (Batch and BatchProcess) –

# batch.py
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
        Params:
            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
        """
        self.batch_cmds.append({
            '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
        self.set_status('STARTED')

        logger.info('Batch:\n{0}'.format(self.batch_cmds))
        for next_cmd in self.batch_cmds:
            logger.info('CWD=={0}'.format(next_cmd['cwd']))
            logger.info('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(
                cmd,
                cwd='{0}'.format(next_cmd['cwd']),
                stderr=subprocess.STDOUT,
                stdout=subprocess.PIPE)
            next_cmd['stdout'] = proc.communicate()[0]
            stdout = next_cmd['stdout'].decode('string_escape')
            logger.info('stdout:\n{0}'.format(stdout))
            exit_code = proc.returncode

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

        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 helion_cli.py (HelionCliComposer class) –

#helicon_cli.py
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
        pass

    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',
            '{0}'.format(self.username),
            '--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),
            '--no-prompt']

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

See package.py (Package class) –

# package.py
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
        Params:
            package_id: package id or name
            package_path: full path of the package (including file name)
        """
        self.id = package_id
        self.file_name = os.path.basename(package_path)
        self.name = self.get_package_name(package_path)
        self.path = os.path.abspath(os.path.expanduser(package_path))
        self.destination = self.get_destination(endpoint_url, self.name)
        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 = regex.search(endpoint_url.strip('/'))
            if (re_match is not None):
                prot = re_match.group(1)
                addr = re_match.group(2)
                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(self.name)

    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 deploy_status.py (DeploymentStatus class) –

# deploy_status.py
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 package.name
        self.store = store

    def get_all(self):
        return self.store.get_deployment_list()

    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}
        try:
            filename = self.get_deployment_filename(id)
            contents = self.store.get_file_contents(filename)

            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)
        """
        logger.info('======= 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):
            history.append(record)
        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)
        self.store.save_file(container_name, filename, contents)

        return result

See deploy.py (Deployment class) –

# deploy.py
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__(
            self,
            package, cli_composer, deploy_status,
            use_package_path=False):
        """
        Initialize an instance of Deployment
        """
        import uuid
        if (use_package_path):
            self.batch = Batch(package.cwd)
        else:
            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
        self.store = deploy_status.store  # backend store
        self.started = False

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

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

        self.get_deployment_batch()

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

            logger.info('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.'
            logger.exception(err_message)
            self.set_status('FAILED')
        else:
            logger.info('DONE deployment - {0}'.format(self.deployment_id))
        finally:
            self.cleanup()

    def download_package(self):
        if (self.cwd_use_package_path):
            self.set_status('DOWNLOADING')
            pkg_filename = self.package.file_name
            pkg_contents = store.get_file_contents(pkg_filename)
            logger.info('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
                package_file.write(pkg_contents)
        return self.package.path

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

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

        # 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())
        self.batch.add(
            '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 = self.package.name

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

        view_cmd = [
            'tar', '-tvf',
            '{0}'.format(src_path)]
        # Assume any foo.tar.gz contains -
        #   - foo/foo.tar.gz (the package to deploy)
        #   - manifest.json
        unpack_cmd = [
            'tar', '-zxvf',
            '{0}'.format(src_path)]
        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
        pass

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

See deployment.py (app module) –

# deployment.py
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)
    Params:
        package_name - the name of the package to deploy
        endpoint_url - the destination (e.g. ALS/Cloud Foundry) endpoint URL
                       ie: 'https://api.15.126.129.33.xip.io'
        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 = ''
    try:
        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

        deployment.set_status('INIT')

        # Start a new process to execute the deployment
        process = Process(
            name='deployment_{0}'.format(deployment_id),
            target=deployment.deploy)
        process.start()

        logger.info('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)
        logger.exception(error_message)
        delete_directory_tree(cwd)
        return {'status': 500, 'errors': error_message}


def get_status(id):
    """
    Get the deployment status by id
    """
    try:
        logger.info("======= 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}
        else:
            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)
        logger.exception(error_message)
        return {'status': 500, 'errors': error_message}

Python class vs module

with one comment


In object-oriented programming design, using class module is one solution to have dependencies injected and encapsulated in constructor (class initialization) with all public methods defining/implementing clear interfaces (contract). Although Python is not a typical OO language (nor a preference many developers using that way), it has class mechanism to accomplish such task. Let’s take a first look on python class.

1. Python class mechanism

The following demo is to write a swiftclient wrapper to access object store on OpenStack (cloud computing platform). Since swift client API methods requires certain dependencies (e.g. auth token, swift URL, and container name), it would be nice to wrap them in a class object instead of calling API methods with these in parameters. For example, ideal coding in application domain should look like this –

from keystone import get_auth_token
from swift import get_swift, Swift

# Use built-in swift_class module to get an instance of Swift
swift = get_swift()

# Otherwise, if dependencies (e.g. swift_url and container_name) in app domain
auth_token = get_auth_token()
swift = Swift(auth_token, swift_url, container_name)

# Once we have a Swift instance ...
afile = swift.get_file_contents('foo.txt')
files = swift.get_files_in_container()

Notice that if application settings need to be separated from swift module (most likely in a large-scale project), the class instantiation will be in application domain (where e.g. settings belong to); otherwise, the swift module could have a get_swift() function to construct a swift instance (like in this demo). The class module and unit tests are listed as below.


See common/swift.py (Swift class) –

# swift.py (Swift class)
import logging
import mimetypes
import sys

from swiftclient import client as swift_client
from swiftclient.exceptions import ClientException
from keystone import get_auth_token
from config import settings

logger = logging.getLogger(__name__)


# ================================================
# Swift instance initialization
# ================================================
def get_swift():
    """
    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


# ================================================
# Swift class
# ================================================
class Swift(object):
    def __init__(self, auth_token, swift_url, container_name):
        """
        Initialize a Swift instance
        """
        self.auth_token = auth_token
        self.swift_url = swift_url
        self.container = container_name
        self.connection = self._get_connection()

    def _get_connection(self):
        try:
            return swift_client.Connection(
                preauthurl=self.swift_url,
                preauthtoken=self.auth_token,
                retries=5,
                auth_version='1',
                insecure=True)
        except Exception as e:
            err_message = "Exception raised initiating a swift connection."
            logger.exception(err_message)
            raise

    def check_container(self):
        """
        Determine if default container missing in Swift
        """
        try:
            headers, container_list = self.connection.get_account()
            for container in container_list:
                if container['name'] == self.container:
                    return True
            return False
        except Exception as e:
            err_message = "Exception raised on checking container exists."
            logger.exception(err_message)
            raise

    def ensure_container_exists(self):
        """
        Ensure default container exists in Swift.
        """
        # Determine if necessary container missing; if so, create it
        container_exists = self.check_container()
        if (not container_exists):
            try:
                response = {}
                self.connection.put_container(
                    self.container, response_dict=response)
            except Exception as e:
                err = "Exception on creating container {0}.".format(
                    self.container)
                logger.exception(err)
                raise

    def get_file_contents(self, file_name):
        """
        Function wrapper to perform 'get_object' call on Swift
        """
        try:
            response_dict = {}
            # Response from Swift:
            #     a tuple of (response headers, the object contents)
            #     The response headers will be a dict and all header names
            #     will be lowercase.
            response = self.connection.get_object(
                self.container,
                file_name,
                response_dict=response_dict)
            file_contents = response[1]
            return file_contents
        except Exception as e:
            err = "Exception on getting {0} from Swift.".format(file_name)
            logger.exception(err)
            raise

    def get_files_in_container(self):
        result = self.connection.get_container(
            self.container, full_listing=True)
        return result[1]

    def save_file(self, file_name, file_contents):
        """
        Function wrapper to perform 'put_object' call Swift
        """
        try:
            self.ensure_container_exists()
            response = {}
            # Example of response from put_object call -
            # {
            #     'status': 201,
            #     'headers': {
            #         'content-length': '0',
            #         'last-modified': 'Fri, 17 Jul 2015 04:43:56 GMT',
            #         'connection': 'keep-alive',
            #         'etag': 'd41d8cd98f00b204e9800998ecf8427e',
            #         'x-trans-id': 'txeddbca07d8e744deae343-0055a8880c',
            #         'date': 'Fri, 17 Jul 2015 04:43:57 GMT',
            #         'content-type': 'text/html; charset=UTF-8'},
            #     'reason': 'Created',
            #     'response_dicts': [{
            #         'status': 201,
            #         'headers': {
            #             'content-length': '0',
            #             'last-modified':
            #             'Fri, 17 Jul 2015 04:43:56 GMT',
            #             'connection': 'keep-alive',
            #             'etag': 'd41d8cd98f00b204e9800998ecf8427e',
            #             'x-trans-id': 'txeddbca07d8e744deae343-0055a8880c',
            #             'date': 'Fri, 17 Jul 2015 04:43:57 GMT',
            #             'content-type': 'text/html; charset=UTF-8'},
            #             'reason': 'Created'}]}
            self.connection.put_object(
                self.container,
                file_name,
                file_contents,
                content_length=sys.getsizeof(file_contents),
                content_type=mimetypes.guess_type(file_name, strict=True)[0],
                response_dict=response)
            return response
        except Exception as e:
            err_message = "Exception on saving file contents to Swift.\n"
            logger.exception(err_message)
            raise


See common/tests/swift_tests.py (Unit tests for Swift class) –

# swift_tests.py
import logging
import mock
import os
import StringIO
import sys
import tarfile
import unittest

from pyramid import testing
from mock import Mock, MagicMock, patch, mock_open
from swiftclient.exceptions import ClientException

from swift import Swift
logger = logging.getLogger(__name__)


class SwiftClassTests(unittest.TestCase):
    @patch('swiftclient.client.Connection')
    def setUp(self, mock_connection):
        self.config = testing.setUp()
        self.auth_token = Mock()
        self.account_data = ([{}], [{'name': 'container1'}])
        self.container = 'container1'
        self.container_data = ([{}], [{'name': 'file1'}, {'name': 'file2'}])
        self.swift_url = 'http://0.0.0.0'
        self.connection = Mock()
        # patch('common.swift.Swift.get_connection',
        #       return_value=self.connection
        mock_connection.return_value = self.connection
        self.swift = Swift(
            self.auth_token, self.swift_url, self.container)

    def tearDown(self):
        testing.tearDown()

    @patch('swiftclient.client.Connection')
    def test_constructor(self, mock_connection):
        self.assertEqual(self.swift.auth_token, self.auth_token)
        self.assertEqual(self.swift.swift_url, self.swift_url)
        self.assertEqual(self.swift.connection, self.connection)
        self.assertEqual(self.swift.container, self.container)

    @patch('swiftclient.client.Connection')
    def test_connection_exception(self, mock_connection):
        error_message = 'CONNECTION ERROR'
        mock_connection.side_effect = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            result = Swift(self.auth_token, self.swift_url, self.container)
            self.assertEqual(str(cm.exception), error_message)

    def test_check_container_exception(self):
        name = 'foo'
        error_message = 'PUT CONTAINER ERROR'
        self.swift.connection.get_account.side_effect \
            = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            result = self.swift.check_container()
            self.assertEqual(str(cm.exception), error_message)
            self.assertFalse(result)

    def test_check_container_when_false(self):
        self.swift.connection.get_account.return_value = self.account_data
        self.swift.container = 'dummy'
        result = self.swift.check_container()
        self.assertFalse(result)

    def test_check_container_when_true(self):
        self.swift.connection.get_account.return_value = self.account_data
        result = self.swift.check_container()
        self.assertTrue(result)

    def test_check_file_exists(self):
        self.swift.check_container = MagicMock()
        self.swift.check_container.return_value = True
        self.swift.connection.get_container = MagicMock()
        self.swift.connection.get_container.return_value = self.container_data
        result = self.swift.check_file_exists('fileX')
        self.assertFalse(result)
        result = self.swift.check_file_exists('file1')
        self.assertTrue(result)

    def test_check_file_exists_no_container(self):
        self.swift.check_container = MagicMock()
        self.swift.check_container.return_value = False
        result = self.swift.check_file_exists('filename')
        self.assertFalse(result)

    def test_ensure_container_exists(self):
        success = {'status': 200}

        def mock_put_success(container_name, response_dict):
            logger.debug(
                'Called with {0} {1}'.format(container_name, response_dict))
            response_dict['status'] = success['status']

        with patch.object(
                self.swift.connection, 'get_account',
                return_value=self.account_data):
            with patch.object(
                    self.swift.connection, 'put_container',
                    side_effect=mock_put_success) as mocked_put:
                self.swift.ensure_container_exists()
                mocked_put.assert_called()

    def test_ensure_container_exists_exception(self):
        error_message = 'PUT CONTAINER ERROR'

        with patch.object(
                self.swift.connection, 'get_account',
                return_value=self.account_data):
            with patch.object(
                    self.swift.connection, 'put_container',
                    side_effect=Exception(error_message)) as mocked_put:
                self.swift.check_container = MagicMock()
                self.swift.check_container.return_value = False
                with self.assertRaises(Exception) as cm:
                    self.swift.ensure_container_exists()
                    self.assertEqual(str(cm.exception), error_message)

    def test_get_file_contents(self):
        response = ([{}], '_filecontents')
        self.swift.connection.get_object = MagicMock()
        self.swift.connection.get_object.return_value = response
        result = self.swift.get_file_contents('file_name')
        self.assertEqual(result, response[1])

    def test_get_file_contents_exeption(self):
        error_message = 'Exception on get object'
        self.swift.connection.get_object = MagicMock()
        self.swift.connection.get_object.side_effect = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            result = self.swift.get_file_contents('file_name')
            self.assertEqual(str(cm.exception), error_message)

    def test_get_files_in_container(self):
        self.swift.connection.get_container = MagicMock()
        self.swift.connection.get_container.return_value = self.container_data
        result = self.swift.get_files_in_container()
        self.assertEqual(result, self.container_data[1])

    @patch('mimetypes.guess_type')
    @patch('sys.getsizeof')
    def test_save_file_contents(self, mock_getsizeof, mock_guess_type):
        success = {'status': 200}

        def mock_put_success(
                container_name, file_name, contents,
                content_length, content_type, response_dict):
            response_dict['status'] = success['status']

        file_name = 'filename'
        contents = MagicMock()
        mock_getsizeof.return_value = 999
        mock_guess_type.return_value = ['filetype']
        with mock.patch.object(
                self.swift.connection, 'put_object',
                side_effect=mock_put_success) as mocked_put:
            self.swift.check_container = MagicMock()
            self.swift.check_container.return_value = True
            self.swift.save_file_contents(file_name, contents)
            mocked_put.assert_called_with(
                self.container,
                file_name, contents,
                content_length=999, content_type='filetype',
                response_dict=success)

    @patch('mimetypes.guess_type')
    @patch('sys.getsizeof')
    def test_save_file_contents_Exception(
            self, mock_getsizeof, mock_guess_type):
        file_name = 'filename'
        contents = MagicMock()
        mock_getsizeof.return_value = 999
        mock_guess_type.return_value = ['filetype']
        error_message = "SWIFT PUT OBJECT ERROR"
        self.swift.connection.put_object.side_effect = Exception(error_message)
        self.swift.check_container = MagicMock()
        self.swift.check_container.return_value = False
        with self.assertRaises(Exception) as cm:
            self.swift.save_file_contents(file_name, contents)
            self.assertEqual(str(cm.exception), error_message)

2. Python module

In use of Python class mechanism (as in above example), it takes a couple steps (of initialization, if no dependency injection used) before all class methods are accessible. Next, let’s see a comparison on how to make swift functions (more in a traditional python style) available right after import the module, in order to have fewer lines (under certain conditions, see discussion on next) of code as demonstrated below. The full source code (with config, keystone modules, and unit tests) are included at the end.

Beware of how @checks_config decorator is used, in swift_module.py, to guarantee properly instantiating a SwiftConfig if any swift method is called without one. One benefit using optional config=None parameter is to have some flexibility that a config can be either specified by the caller or created by swift module automatically. This makes swift module methods have same signatures (without config), comparing to class mechanism. But disadvantage (of such hiding) also appears that each auto-creation of a config by the decorator will have a different instance.

import swift_module as swift

# optionally to get a swift config first (see SwiftConfig in swift_module.py)
config = swift.get_swift_config()
# swift module functions are available after the import
a_file = swift.get_file_contents('foo.txt', config=config)
allist = swift.get_files_in_container(config=config)


See common/config.py

import logging
import pyramid

logger = logging.getLogger(__name__)


def checks_config(config_func):
    """
    Get decorator to use config_func as initiator for 'config' arg

    Keyword arguments:
    config_func -- a function to get proper configuration
    """
    def checks_config_decorator(original_func):
        """
        Call decorated original_func with checking its 'config' arg

        Keyword arguments:
        func -- original function to be decorated
        """
        def _arg_index_of(func, name):
            """
            Get the index of a named arg on a func call
            """
            import inspect
            argspec = inspect.getargspec(func)
            for i in range(len(argspec[0])):
                if (argspec[0][i] == name):
                    logger.debug("argspec[0][{0}]=={1}".format(i, name))
                    return i
            return -1

        def _checks_config_wrapper(*args, **kwargs):
            """
            Check 'config' arg before calling original_func
            Call specified config_func if 'config' arg value is None.
            """
            arg_idx = _arg_index_of(original_func, 'config')
            has_arg = 0 <= arg_idx and arg_idx < len(args)
            arg_cfg = args[arg_idx] if (has_arg) else None
            kwa_cfg = kwargs.get('config')
            if (kwa_cfg is None and arg_cfg is None):
                # logger.debug('Getting config by {0}'.format(config_func))
                kwargs['config'] = config_func()
            return original_func(*args, **kwargs)

        # calls the original function with checking proper configuration
        return _checks_config_wrapper
    # returns a decorated function
    return checks_config_decorator


def settings(item):
    """
    Get a reference to the settings in the .ini file
    """
    registry = pyramid.threadlocal.get_current_registry()
    return registry.settings.get(item, None)


See common/keystone.py

from config import settings
from keystoneclient.v2_0 import client as keystone_client
from logging import getLogger
logger = getLogger(__name__)


def get_auth_token():
    """
    Get an auth token from Keystone.
    """
    try:
        keystone = keystone_client.Client(
            username=settings('cloud_username'),
            password=settings('cloud_password'),
            tenant_name=settings('cloud_project_name'),
            auth_url=settings('cloud_auth_url'),
            insecure=True)
        return keystone.auth_ref['token']['id']
    except Exception as e:
        logger.error(
            "Exception authenticating against Keystone")
        logger.exception("Details: {0}".format(e))
        raise


See common/swift_module.py

# swift_module.py
import logging
import mimetypes
import sys

from config import checks_config, settings
from keystone import get_auth_token
from swiftclient import client as swift_client
from swiftclient.exceptions import ClientException

logger = logging.getLogger(__name__)


# ================================================
# Swift configuration class
# ================================================
class SwiftConfig(object):
    def __init__(self, auth_token, swift_url, container_name):
        """
        Initialize a Swift configuration instance
        """
        self.auth_token = auth_token
        self.swift_url = swift_url
        self.container = container_name
        self.connection = self._get_connection()

    def _get_connection(self):
        """
        Get a connection to Swift object store
        """
        try:
            return swift_client.Connection(
                preauthurl=self.swift_url,
                preauthtoken=self.auth_token,
                retries=5,
                auth_version='1',
                insecure=True)
        except Exception as e:
            err_message = "Exception raised initiating a swift connection."
            logger.exception(err_message)
            raise

# ToDo [zhuyux]: considering singleton for SwiftConfig instance
_swift_config_singleton = None

# ================================================
# Swift configuration initialization
# ================================================
def get_swift_config():
    """
    Get a SwiftConfig instance per application settings
    """
    auth_token = get_auth_token()
    container = settings('swift_container')
    swift_url = settings('swift_url')
    swift_cfg = SwiftConfig(auth_token, swift_url, container)
    return swift_cfg


def _get_config():
    """
    This is a fixed/non-mockable func pointer for @checks_config decorator
    """
    # logger.debug('get_swift_config={0}'.format(get_swift_config))
    return get_swift_config()

# ================================================
# Swift module interfaces
# ================================================

@checks_config(config_func=_get_config)
def check_container(config=None):
    """
    Check if default container exists in Swift

    Keyword arguments:
    config -- an instance of SwiftConfig (optional, default None)
    """
    try:
        logger.debug('Checking container {0}'.format(config.container))
        headers, container_list = config.connection.get_account()
        for container in container_list:
            logger.debug("--- container: {0}".format(container['name']))
            if (container['name'] == config.container):
                logger.debug('--- found {0}'.format(config.container))
                return False
        logger.debug('--- missing container {0}'.format(config.container))
        return True
    except Exception as e:
        err_message = "Exception raised on checking container exists."
        logger.exception(err_message)
        raise


@checks_config(config_func=_get_config)
def check_file_exists(file_name, config=None):
    """
    Check if specified file exists in Swift store

    Keyword arguments:
    file_name -- the name of the file to be checked in Swift store
    config -- an instance of SwiftConfig (optional, default None)
    """
    if (check_container(config=config)):
        files = get_files_in_container(config=config)
        for file in files:
            if (file['name'] == file_name):
                return True
    return False


@checks_config(config_func=_get_config)
def ensure_container_exists(config=None):
    """
    Ensure default container exists in Swift.

    Keyword arguments:
    config -- an instance of SwiftConfig (optional, default None)
    """
    container_exists = check_container(config=config)
    if (not container_exists):
        try:
            response = {}
            config.connection.put_container(
                config.container, response_dict=response)
            logger.debug(
                "--- Container {0} created".format(config.container))
            logger.debug("--- Response {0}".format(response))
        except Exception as e:
            err = "Exception on creating container {0}.".format(
                config.container)
            logger.exception(err)
            raise


@checks_config(config_func=_get_config)
def get_file_contents(file_name, config=None):
    """
    Function wrapper to perform 'get_object' call on Swift

    Keyword arguments:
    file_name -- the name of the file in Swift store
    config -- an instance of SwiftConfig (optional, default None)
    """
    try:
        response_dict = {}
        # Response from Swift:
        #   a tuple of (response headers, the object contents)
        #   The response headers will be a dict and all header names
        #   will be in lower case.
        response = config.connection.get_object(
            config.container,
            file_name,
            response_dict=response_dict)
        file_contents = response[1]
        return file_contents
    except Exception as e:
        err = "Exception on getting {0} from Swift.".format(file_name)
        logger.exception(err)
        raise


@checks_config(config_func=_get_config)
def get_files_in_container(config=None):
    """
    Get info of all files in default Swift container

    Keyword arguments:
    config -- an instance of SwiftConfig (optional, default None)
    """
    result = config.connection.get_container(
        config.container, full_listing=True)
    return result[1]


@checks_config(config_func=_get_config)
def save_file(file_name, file_contents, config=None):
    """
    Function wrapper to perform 'put_object' call Swift

    Keyword arguments:
    file_name -- the name of the file to be saved
    file_contents -- the contents of the file to be saved in Swift store
    config -- an instance of SwiftConfig (optional, default None)
    """
    try:
        # Ensure the default container exists
        ensure_container_exists(config=config)
        # Push the file contents to Swift
        response = {}
        config.connection.put_object(
            config.container,
            file_name,
            file_contents,
            content_length=sys.getsizeof(file_contents),
            content_type=mimetypes.guess_type(file_name, strict=True)[0],
            response_dict=response)
        return response
    except Exception as e:
        err = "Exception on saving file contents to Swift.\n"
        logger.exception(err)
        raise


See common/tests/swift_module_tests.py

import logging
import mock
import os
import StringIO
import sys
import tarfile
import unittest

import common.swift_module as swift

from pyramid import testing
from mock import Mock, MagicMock, patch, mock_open
from swiftclient.exceptions import ClientException

logger = logging.getLogger(__name__)


class SwiftTests(unittest.TestCase):
    @patch('common.swift_module.get_auth_token')
    @patch('swiftclient.client.Connection')
    def setUp(self, mock_swift_connection, mock_get_auth_token):
        self.config = testing.setUp()
        self.auth_token = MagicMock()
        self.account_data = ([{}], [{'name': 'container1'}])
        self.container = 'container1'
        self.container_data = ([{}], [{'name': 'file1'}, {'name': 'file2'}])
        self.swift_url = 'http://0.0.0.0'
        self.setting = 'dummy setting'
        self.connection = Mock()
        mock_swift_connection.return_value = self.connection
        mock_get_auth_token.return_value = self.auth_token

        self.swift_cfg = swift.SwiftConfig(
            self.auth_token, self.swift_url, self.container)
        self.swift_cfg.connection.get_account.return_value = self.account_data

    def tearDown(self):
        testing.tearDown()

    @patch('swiftclient.client.Connection')
    def test_swift_config(self, mock_connection):
        self.assertEqual(self.swift_cfg.auth_token, self.auth_token)
        self.assertEqual(self.swift_cfg.swift_url, self.swift_url)
        self.assertEqual(self.swift_cfg.connection, self.connection)
        self.assertEqual(self.swift_cfg.container, self.container)

    @patch('swiftclient.client.Connection')
    def test_swift_config_exception(self, mock_connection):
        error_message = 'CONNECTION ERROR'
        mock_connection.side_effect = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            result = swift.SwiftConfig(
                self.auth_token, self.swift_url, self.container)
            self.assertEqual(str(cm.exception), error_message)

    def test_check_container_exception(self):
        error_message = 'GET ACCOUNT ERROR'
        self.swift_cfg.connection.get_account.side_effect \
            = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            result = swift.check_container(config=self.swift_cfg)
            self.assertEqual(str(cm.exception), error_message)
            self.assertFalse(result)

    @patch('common.swift_module.get_swift_config')
    def test_check_container_when_false(self, mock_get_config):
        mock_get_config.return_value = self.swift_cfg
        self.swift_cfg.container = '-=#=-dummy-=#=-'
        result = swift.check_container()
        self.assertFalse(result)

    @patch('common.swift_module.get_auth_token')
    @patch('swiftclient.client.Connection')
    def test_check_container_when_true(
            self, mock_swift_connection, mock_get_auth_token):
        mock_swift_connection.return_value = self.connection
        mock_get_auth_token.return_value = self.auth_token
        result = swift.check_container(config=self.swift_cfg)
        self.assertTrue(result)
        self.swift_cfg.container = '-=#=-dummy-=#=-'
        result = swift.check_container(config=self.swift_cfg)
        self.assertFalse(result)

    @patch('common.swift_module.get_swift_config')
    @patch('common.swift_module.check_container')
    def test_check_file_exists(self, mock_check_container, mock_get_config):
        mock_check_container.return_value = True
        mock_get_config.return_value = self.swift_cfg
        self.swift_cfg.connection.get_container = MagicMock()
        self.swift_cfg.connection.get_container.return_value = \
            self.container_data
        result = swift.check_file_exists('fileX')
        self.assertFalse(result)
        result = swift.check_file_exists('file1')
        self.assertTrue(result)

    @patch('common.swift_module.check_container')
    def test_check_file_exists_no_container(self, mock_check_container):
        mock_check_container.return_value = False
        result = swift.check_file_exists('filename', config=self.swift_cfg)
        self.assertFalse(result)

    def test_ensure_container_exists(self):
        success = {'status': 200}

        def mock_put_success(container_name, response_dict):
            logger.debug(
                'Called with {0} {1}'.format(container_name, response_dict))
            response_dict['status'] = success['status']

        with patch.object(
                self.swift_cfg.connection, 'get_account',
                return_value=self.account_data):
            with patch.object(
                    self.swift_cfg.connection, 'put_container',
                    side_effect=mock_put_success) as mocked_put:
                swift.ensure_container_exists(config=self.swift_cfg)
                mocked_put.assert_called()

    def test_ensure_container_exists_exception(self):
        error_message = 'PUT CONTAINER ERROR'

        with patch.object(
                self.swift_cfg.connection, 'get_account',
                return_value=self.account_data):
            with patch.object(
                    self.swift_cfg.connection, 'put_container',
                    side_effect=Exception(error_message)) as mocked_put:
                with self.assertRaises(Exception) as cm:
                    import common.swift
                    swift.ensure_container_exists(config=self.swift_cfg)
                    self.assertEqual(str(cm.exception), error_message)

    def test_get_file_contents(self):
        import common.swift
        response = ([{}], '_filecontents')
        self.swift_cfg.connection.get_object = MagicMock()
        self.swift_cfg.connection.get_object.return_value = response
        result = swift.get_file_contents('file_name', config=self.swift_cfg)
        self.assertEqual(result, response[1])

    def test_get_file_contents_exeption(self):
        error_message = 'Exception on get object'
        self.swift_cfg.connection.get_object = MagicMock()
        self.swift_cfg.connection.get_object.side_effect = Exception(error_message)
        with self.assertRaises(Exception) as cm:
            swift.get_file_contents('file_name', config=self.swift_cfg)
            self.assertEqual(str(cm.exception), error_message)

    def test_get_files_in_container(self):
        self.swift_cfg.connection.get_container = MagicMock()
        self.swift_cfg.connection.get_container.return_value = \
            self.container_data
        result = swift.get_files_in_container(config=self.swift_cfg)
        self.assertEqual(result, self.container_data[1])

    @patch('common.config.settings')
    @patch('common.swift_module.get_auth_token')
    @patch('swiftclient.client.Connection')
    def test_get_swift_config(
            self, mock_connection, mock_get_auth_token, mock_settings):
        mock_connection.return_value = self.connection
        mock_get_auth_token.return_value = self.auth_token
        mock_settings.return_value = self.setting
        result = swift.get_swift_config()
        self.assertEqual(result.auth_token, self.auth_token)
        self.assertEqual(result.connection, self.connection)
        self.assertEqual(result.container, self.setting)
        self.assertEqual(result.swift_url, self.setting)

    @patch('mimetypes.guess_type')
    @patch('sys.getsizeof')
    def test_save_file(self, mock_getsizeof, mock_guess_type):
        success = {'status': 200}

        def mock_put_success(
                container_name, file_name, contents,
                content_length, content_type, response_dict):
            response_dict['status'] = success['status']

        mock_getsizeof.return_value = 999
        mock_guess_type.return_value = ['filetype']
        with mock.patch.object(
                self.swift_cfg.connection, 'put_object',
                side_effect=mock_put_success) as mocked_put:
            import common.swift
            swift.check_container = MagicMock()
            swift.check_container.return_value = True
            filename = 'filename'
            contents = MagicMock()
            swift.save_file(filename, contents, config=self.swift_cfg)
            mocked_put.assert_called_with(
                self.container,
                filename, contents,
                content_length=999, content_type='filetype',
                response_dict=success)

    @patch('mimetypes.guess_type')
    @patch('sys.getsizeof')
    def test_save_file_Exception(
            self, mock_getsizeof, mock_guess_type):
        import common.swift
        mock_getsizeof.return_value = 999
        mock_guess_type.return_value = ['filetype']
        error_message = "SWIFT PUT OBJECT ERROR"
        self.swift_cfg.connection.put_object.side_effect = \
            Exception(error_message)
        swift.check_container = MagicMock()
        swift.check_container.return_value = False
        with self.assertRaises(Exception) as cm:
            filename = 'filename'
            contents = MagicMock()
            swift.save_file(filename, contents, config=self.swift_cfg)
            self.assertEqual(str(cm.exception), error_message)

Summary

Programmers like to write simple code (to make it clear and easy to understand, thus easier to share, test, and maintain). Sure there is always a cost of effort. In OO design (either with C++, Java, or C#), dependency injection pattern has been applied, not only to write less and cleaner code, but also to support the concept of “programming to interfaces, not implementations” – forcing us to honor the contracts.

Python began as a C-like scripting language. Its class mechanism and dependency injection are not widely adopted. However, good design pattern concepts should apply on large scale projects. The demo above has given a comparison of how to program to interfaces in two different ways. Since the class demo does not use any dependency injection, it has fewer lines of code as in swift class module, but more lines on class initialization.

On the other hand, the python module way has encapsulated all dependencies in swift module, all methods become immediately accessible after the import. This is not the best example to prove using “traditional” python module is better. Because, if there are more dependencies coming from different domains or tiers, it would be difficult to achieve the same result (without writing a lot of wrappers or decorators).

If any module method has simple interface and dependencies, python module is just working fine (plus a little complication on config in this demo); otherwise, if it ends up requiring many parameters (or dependencies), python class should be in a design consideration – after all, there must be a bigger reason of introducing class into python than just pleasing OO developers.

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.

Quotes from Winnie-the-Pooh

leave a comment »


Quotes from Winnie-the-Pooh (A.A. Milne)

  • If you live to be a hundred, I want to live to be a hundred minus one day so I never have to live without you.
  • I’m not lost for I know where I am. But however, where I am may be lost.
  • I think we dream so we don’t have to be apart for so long. If we’re in each other’s dreams, we can be together all the time.
  • People say nothing is impossible, but I do nothing every day. (see this)
  • Rivers know this: there is no hurry. We shall get there some day.
  • Some people care too much. I think it’s called love.

Written by Boathill

2015-05-16 at 13:00

Posted in digest, quote

Tagged with , ,

Big 4 Ice Caves

leave a comment »

Written by Boathill

2015-05-09 at 09:00

Bao-pu-zi

leave a comment »


东晋葛洪语录(《抱朴子》名句摘抄)

★、学之广在于不倦,不倦在于固志。
  见晋·葛洪《抱朴子·崇教》。固志:坚定志向。这两句大意是:学问的广博在于学而不倦,学而不倦在于志向坚定。二句严密的逻辑,指出学问的博大精深来源于学而不倦,能作到学而不倦的内因在于目的纯正,志向坚定。苟子《劝学篇》说:“不积跬步,无以至千里;不枳小流,无以成江海。骐骥一跃,不能十步;驽马十驾,功在不舍”。可见学问在于积累,而积累必须有恒心,有志者事竟成。可以这两句说明青年人应树立正确的学习目的和坚定的志向,才能有强大的学习动力,才会作出巨大的成就,

★、用得其长,则才无或弃;偏诘其短,则触物无可。
  晋·葛洪《抱朴子·博喻》用人如果用他擅长的方面,那么人才就不会被弃之不用;如果片面责难他不擅长的方面,那么所有的人都会觉得不合意。出自《抱朴子·博喻》。诘(jié):追问,责问。

★、疏广散金以除子孙之祸。
  晋·葛洪《抱朴子·守塉》。疏广:西汉宣帝时人,任太子太傅五年,后称病还乡。把皇帝所赐黄金广济乡人。他认为把钱财留给子孙,其子孙“贤而多财,则损其志;愚而多财,刚益其过”。本句大意是.疏广把自己的钱财散发给乡里邻居,为的是消除子孙后代未来的灾祸。疏广所为,不失为远见卓识之举。那些倚仗自己的权势、地位和优裕条件极力为子孙谋利益的人,对子孙看似爱之,实则害之,即使留下遗产如金山银山,不知道自力更生的不肖子孙也会坐吃山空,挥霍净尽,甚至为非作歹,作奸犯科,作父母的恐怕不能辞其咎吧。

★、明治病之术者,杜未生之疾。
  晋·葛洪《抱朴子·用刑》。杜:断绝。这两句大意是:深明病理和医术的人,可以杜绝没有发生的疾病。医术高明的医生,深明病理,了解疾病产生的原因,可以在疾病还没暴露出来之前先作预防,不使疾病产生,可用于说明良医不单能治疗疾病,而且能预防疾病;也可用以比喻为政和处世经验丰富的人,可以预防、杜绝事故的发生。

★、官达者,才未必当其位;誊美者,实未必副其名。
  晋·葛洪《抱朴子·博喻》。选:显贵。当(dang荡):适台。誉美者:被赞誉称美的人。副:符合。这两句大意是官职显贵的人,才能未必就适合他所占据的位置;设赞誉称美的人,他的行为未必与名声相符。古人立身道德讲究“名实相符”,但在封建社会里的十普遍现象是位高才下,名实不符。可用于对这种社会现象的讽刺,也可用以提醒人们对“官达”、“誉美”者要考察其实际表现不可因其地位、名声而轻信。

★、水则不决不流,不积不深。
  晋·葛洪《抱朴子·勖学》。决:疏导,开通水道。这两句大意是:水道不疏通,水就不会畅通奔流;水若不积蓄容汇,也就不会成为深广的江海。这两句以水为比喻,前句说明在学习中遇有疑难,困惑不解,如果没有人具体指导,解疑释难,就不会茅塞顿开,一解百解;后句说明学习是一个渐进过程,如果不能坚持不懈,日积月累,就不能成为知识渊博、有学问有本领的人。

★、火则不钻不生,不扇不炽。
  晋·葛洪《抱朴子·勖学》。钻:指钻木生火。炽:炽烈,火旺。这两句大意是:不去钻木就生不出火来,不用扇煽,火苗就不会炽烈旺盛。比喻任何事物的产生和发展,都必须具备相应的外因。因此,要搞发明创造,要促进事业的兴旺发达,必须先创造必不可少的条件,否则只能成为空谈。

  ★、金以刚折,水以柔全,山以高陊,谷以卑安。
晋·葛洪《抱朴子·广譬》。以:因为。全:保全。陊(duò堕):坠落。这几句大意是,金属所以易断折,是因为它刚强;水所以能安全,是因为它柔和;高山容易发生山崩而坠落,是因为它高;山谷所以能平安自适,是因为它低。几句以自然物性为喻,阐述一种哲理,一种社会现象:刚强耿直的人,容易受挫折、遭打击;柔和无争的人,则能够明哲保身。“露头的椽子先烂”,“树大招风”,地位高的头面人物,竞争者多,树敌也多,易受攻击,也容易倒台;地位低下的人,默默无闻,不被人所知,也不为人所重,倒常平安无事。

★、名美而实不副者,必无没世之风。
  晋·葛洪《抱朴子·博喻》。副:相称,符合。没世:永久。风:风范。这两句大意是:博得美名而与实际不副的人,必无永久的可以传世的风薄。世界上有许多名不副实的人或物,凭借机遇、势力、欺骗宣传或其他原因,名噪一时,甚或名重一世,但却经不起时间老人的检验,最终必将还其本来面目,这就是历史的辩证法。

★、金舟不能凌阳侯之波,玉马不任骋千里之迹。
  晋·葛洪《抱朴子·用刑》。凌:渡。阳候:古代传说中的波涛之神,此指洪波巨浪。任:胜任。骋:驰骋,奔驰。这两句大意是:金制的船不能横渡洪波巨浪,玉雕的马不能胜任千里驰驱。船的功用在于渡河,马的功用在于奔驰。金舟玉马既贵重又精美,却无渡河之实和驰骋之用,因此徒有“舟”、“马’之名,而无“舟”、“马”之实,只具有可供观赏的工艺品价值而已。~所说明的是成事必须务实的道理,颇有鉴戒意味。

★、千仓万箱,非一耕所得;干天之木,非旬日所长。
  晋·葛洪《抱朴子·极言》。一耕:一次耕耘。干天:冲天,耸人云天。木:树。旬日:十日。长(zhǎng掌):生长。这几句大意是:千仓万箱的粮食,不是一次耕耘就能得到的;高人云天的大树,不是十天工夫就能长成的。这几句话是比喻任何事业的成功都不可能一蹴而就,要想得到硕果,就必须不断努力,坚持不懈地下工夫,因为任何事物的发展都必须经过渐变的过程。可用于说明对待学习和事业应持的态度。

★、小疵不足以损大器,短疾不足以累长才。
  晋·葛洪《抱朴子·博喻》。疵(cī刺):小毛病。本句大意是:不能因为一点儿小毛病就损坏一个大的器物。这句话说明:金无足赤,人无完人,世上任何事物都有缺点,不能吹毛求疵,因小失大,看人待事都应这样。可说明事物的相对性及衡量人才的标准。

★、西施有所恶而不能减其美者,美多也。
  晋·葛洪《抱朴子·博喻》。西施:古代著名的美女。恶:丑。这两句大意是:尽管西施也有长得美中不足的地方,但并不减少她的美色,因为她长得美的地方太多了。俗话说“金无足赤,人无完人”,世上再美好的事物,从不同的角度和视点去观察,也总能找出不尽如人意的地方。但白璧微瑕,并不影响整体的美。这两句可供论述瑕不掩瑜的遭理,也可供论述对任何人和事都不能求全责备的道理。

★、锐锋产乎钝石,明火炽乎暗木,贵珠出乎贱蚌,美玉出乎丑璞。
  晋·葛洪《抱朴子·博喻》。锐锋:指具有锐利锋刃的刀、剑等。钝石:粗笨的矿石。炽(chi赤):原义是火旺,这里是燃烧意。璞(pú仆):中间藏有美玉的石头。这几句大意是:锐利的刀剑来源于粗钝的矿石,明亮的火光来源于燃烧的暗木,珍贵的明珠出自轻贱的河蚌,华美的宝玉出自丑陋的璞石。自来人人都喜爱锐锋、明火、贵珠、美玉,却每每忘了它们来源于不起眼的钝石、暗木、贱蚌、丑璞。这几句可供论述伟大往往孕育于平凡之中的道理,也可供论述从质朴中可以提炼出精美的道理。

★、云厚者,雨必猛,弓劲者,箭必远。
  晋·葛洪《抱朴子·喻蔽》。劲:强劲。这几句大意是:云层密厚的,下的雨一定很猛;硬弓强劲的,射出的箭一定很远。这几句说的虽然是自然现象和物理现象,但其中蕴含的道理却可用于论教和治学,说明博览今古,学贯中西,根基扎实,功底深厚的人,他的见识、成就、学术造诣就会超出一般人。

★、粉黛至则西施以加丽。
  晋·葛洪《抱朴子·勖学》。黛(dài代):青黑色的颜料,古代女子用来画眉。西施:春秋末年越国著名美女。本句大意是:有了化妆品的装饰,越国美女西施会更加美丽。此名句以比喻象征手法说明学习的道理。即使是学富五车的人,也像西施需要粉黛一样地需要学习,孤陋寡闻的人就更不用说了。此句比喻巧妙,寓意深刻,其道理通过具体的形象表现出来,显得既通俗又含蓄,既具体可感又耐人寻味,从而增强了文句本身的艺术魅力。

★、必死之病,不下苦口之药;朽烂之材,不受雕镂之饰。
  晋·葛洪《抱朴子·博喻》。雕镂(lou漏):雕刻。这几句大意是:对于无法挽救的病人,不必再给他下苦口的药了(以免增加他的痛苦);已经朽烂的木头,不能再雕镂精美的花纹了(没有保存的价值)。良药苦口利于病,是对于可治之病讲的,若病入膏肓,无可挽救,还给病人吃苦口之药,只是徒然增加其痛苦罢了;“朽木不可雕也”(见《论语·公冶长》),雕了也没有美学价值,不能成为传世之作。故以此二句说明办事应实事求是,不能沽名钓誉,图幕虚名,否则只是白费功夫,甚至会带来相反的效果。

★、虽云色白,匪染弗丽;虽云味甘,匪和弗美。
  晋·葛洪《抱朴子·勖学》。匪:非。这几句大意是:丝虽然说很白,不经漂染,不会明丽喜人;美味虽然甘甜,不经调和,不会脍炙人口。此条是葛洪在阐明教学的重要性时所打的比方,其目的在于通过未加工与加工过的东西的物性差别,说明教育和学习的必要性。生丝虽然很有价值,然而只有经过了漂染、加工,才能更加明丽。同理.人虽然是万物灵长,也只有经过学习,接受教育,才能成为一个有用的人,发挥出更大的人生价值。此条通过具体可感的实物,说明重视教育的道理,手法巧妙,表述生动。这是古人说理文中常用的一种表现手法。

★、临凝结而能断,操绳墨而无私。
  晋·葛洪《抱朴子·行品》。凝结:指难分难解的纠葛。操:掌握。绳墨:本是木工打直线的工具,比喻规矩或法度。这两句大意是:遇到纠葛而能作出决断,掌握法度而能无所偏私。

★、金钩玉饵虽珍,不能制九渊之沉鳞。
  晋·葛洪《抱朴子·广譬》。饵:钓鱼用的鱼食。九渊:深渊。制:制服,指钧得。沉鳞:沉在水底里的游鱼。这两句大意是:黄金做成鱼钩,白玉当作鱼饵,虽然都租珍贵,却不能钓得沉在渊底的游鱼。天下之物各有所用,只有用得其所,才能发挥它的最大功能,取得最佳的效益;用得不得其所,即使再珍贵,也等于废物。用人也应作如是观。

★、役其所长,则事无废功;避其所短,刚世无弃材。
  晋·葛洪《抱朴子·务正》。役(yì忆):驱使,任用。废功:不成功。材:问“才”。这几句大意是:任用人所擅长的本领,事情就没有不成功的;避开人的短处,世上就没有可弃掉的人才。任用人的长处,因为是其所擅长,干起事来得心应手,就易于把事成功。人有短处,如果任用时能避开他的短处,只用他的长处,帮么世上就班有完全没用的人了。这几句用于说明任用人如果能扬长避短,则人人可用;能人尽其用,也就易于把事情办成功。

★、以玉为石者,亦将以石为玉矣;以贤为愚者,亦将以愚为贤矣。
  晋·葛洪《抱朴子·擢才》。这几句大意是:把宝玉当作石头的人,也会把石头当作宝玉;把贤人当作愚人的人,也会把愚人当作贤人。以前两句比喻后两句,说明一些人知识浅薄,目光短浅,贤愚不分,能把贤人当作愚人,也能把愚人当作贤人。以贤为愚会埋没人才,以愚为贤又会重用蠢才,这对国家对事业都是极其危险的。

★、良骏败于拙御,智士踬于暗世。
  晋·葛洪《抱朴子·官理》。御(yù预):驾驶车马的人。踬(zhì描):被绊倒。这两句大意是:良马失败于拙劣的驾御者,才智之士羁绊于昏暗的世道。好马须有良御,智士须逢明世。再好的马遇到拙劣的骑手和御手也跑不快;再有才智的人生在暗世也发挥不了自己的才能,不能有所作为。这两句以“良骏败于拙御”作比兴,通晓易明地揭示了“智士踬于暗世”的悲剧。这种以浅比深的说理方法可以借鉴。可用这两句揭示智士不能有所作为的原因,也可用以抒发智士逢暗世而不能有所作为的感叹。

★、志合者,不以山海为远;道乖者,不以咫尺为近。故有跋涉而游集,亦或密迩而不接。
  晋·葛洪《抱朴子·博喻》志向投合的人不认为山海的阻隔遥远,意见不合的人不认为咫尺的距离很近。所以有的人跋山涉水从各处来相聚,也有的人就在眼前却不相交往。

★、位高而器不称者,不免致冠之惑也。 虽有兄弟,不如友生。 朋友之交不宜浮杂。

★、坚志者,功名之主也。不惰者,众善之师也。

★、过载者沉其舟,欲胜者杀其身。

★、冬不欲极温,夏不欲穷凉。

Written by Boathill

2015-05-04 at 22:00

Posted in digest

Tagged with , , , , ,

SoCal 2015

leave a comment »

Written by Boathill

2015-01-31 at 22:00

Follow

Get every new post delivered to your Inbox.

%d bloggers like this: