Archive for the ‘IT’ Category
Javascript: Sync JSON Data
/**
* Synchronize new data to original object and keep reference
* Check object data type and traverse object tree to delete old key
* as well as replace with new data:
*
* - if keyed data is directory (hash map) 'object', sync recursively
* - else (including number, string, and array, etc), replace with new data
*
* @returns void.
*/
function syncObjectData(newData, orgData) {
if (newData != null && orgData !== newData) {
var sourceType = orgData.toType();
var targetType = newData.toType();
if (sourceType != targetType) {
// the top level type must match
var message = "Cannot update '" + sourceType + "' with different type " + targetType;
throw new Error(message);
} else if (sourceType != 'object') {
// can only sync object type
throw new Error("Does not support updating type of " + orgType);
} else {
for (var key in newData) {
if (newData.hasOwnProperty(key)) {
if (orgData.hasOwnProperty(key) && newData[key] != null) {
var orgType = toType(orgData[key]);
var newType = toType(newData[key]);
if (newType == orgType && orgType == 'object' && orgData[key] != null) {
syncObjectData(newData[key], orgData[key]);
} else {
orgData[key] = newData[key];
}
} else if (!orgData.hasOwnProperty(key)) {
orgData[key] = newData[key];
} else {
delete orgData[key];
}
}
}
for (var key in orgData) {
if (orgData.hasOwnProperty(key)) {
if (!newData.hasOwnProperty(key) || newData[key] == null) {
delete orgData[key];
}
}
}
}
}
}
Here is the helper function to get name (string) of the exact type.
/**
* Convert data type to string, e.g. 'undefined',
* or 'null', 'boolean', 'number', 'string', 'array', 'object', 'date', 'function', etc.
* https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects
*/
function toType(variable) {
return ({}).toString.call(variable).match(/\s([a-zA-Z]+)/)[1].toLowerCase();
}
Python Class in Design
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
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)
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)
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
# 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.
PicasaWeb full-size download
First of all, since Google+ was introduced, PicasaWeb has been automatically redirecting to plus.google.com, which makes the URL longer and harder to remember, as well as missing some traditional PicasaWeb features/links (for example, “Linked Galleries”, “Sort”, Album Locations, and etc.). Note the transform from PicasaWeb to Google+ albums is from
- https://picasaweb.google.com/jason.zhuyx (where the last part is the email alias)
to
- https://plus.google.com/photos/103192840022832307624/albums (where the digits are Google+ id)
But other Google+ links (e.g. about/profile, posts) are actually starting after the id:
😦
Recently, Google adds a “Click here to go back to Picasa Web Albums” tab (in a light yellow background, for about a minute) over the top of Google+ albums. If the tab disappears, press F5 to refresh the page and get it back. The URL is using a “?noredirect=1” query string to retain the Picasa Web.
Once in an album, each photo opened on PicasaWeb has an image URL (when you use “Save As”, or “Copy Image URL” link from context menu), like this:
However, the above link is only the image rendered in a browser. Depending on the system (e.g. PC, tablet, or phone device) and browser window, the size could be “s512”, “s640”, or “s800” (as the width of the image, or something like “w640-h480-no”, as dynamic rendering size, on Google+ page). For now the width can go up to 2560 (“s2560”) but never be the full size of the original uploaded picture.
In order to download the original photo, just simply change this “size path” to “d” or “s0-d“.
Caution: The changed link will pop up Download dialog instead of rendering the image. DO NOT use the URL in HTML <img src="" />. And hope this protocol stays …
PhotoShop Alternatives
- Aviary (Pad/Phone App)
- GIMP (OpenSource/free on Windows, OS X, Linux) – see user manual and turotials.
- Paint.NET (only for Windows platform)
- LunaPic, or PicMonkey (Browser App)
- PhotoShop Express (aka Photoshop.com, Browser/Pad/Phone App)
- Pixlr (Browser/Pad/Phone App)
See a more complete list at DigiCamHelp and detailed review here.
Other photo edit/view tools:
- Geo Setter (Geo data editor on Windows)
- IrfanView (Compact viewer with basic editing features on Windows)
- Panoramic Factory (see other tools from Ken Rockwell’s review)
- PhotoMatix Pro (see more in this review)
SQL Named Constraint
SQL supports following constraints:
CHECK– Ensures that the value in a column meets a specific conditionDEFAULT– Specifies a default value when specified none for this columnNOT NULL– Indicates that a column cannot store NULL valueFOREIGN KEY– Ensure the referential integrity of the data in one table to match values in another tablePRIMARY KEY– A combination of a NOT NULL and UNIQUE. Ensures that a column (or combination of two or more columns) have an unique identity which helps to find a particular record in a table more easily and quicklyUNIQUE– Ensures that each row for a column must have a unique value
Starting with an example of having named constraint in CREATE statement.
CREATE TABLE [dbo].[Bar]
(
[Id] int CONSTRAINT [CO_Bar_Id_NOTNULL] NOT NULL, -- PRIMARY KEY,
[Name] NVARCHAR(50) -- CONSTRAINT [CO_Bar_Name_NOTNULL] NOT NULL UNIQUE,
--CONSTRAINT [PK_Bar_Id] PRIMARY KEY ([Id]), -- named constraint
CONSTRAINT [CO_Bar_Name_UNIQUE] UNIQUE ([Name]),
)
GO
CREATE TABLE [dbo].[FooType]
(
[Id] int CONSTRAINT [CO_FooType_Id_NOTNULL] NOT NULL, -- PRIMARY KEY,
[Name] NVARCHAR(50) CONSTRAINT [CO_FooType_Name_NOTNULL] NOT NULL UNIQUE,
CONSTRAINT [PK_FooType_Id] PRIMARY KEY ([Id]), -- named constraint
)
GO
CREATE TABLE [dbo].[Foo]
(
[Id] int CONSTRAINT [CO_Foo_Id_NOTNULL] NOT NULL,
[Name] NVARCHAR(50) CONSTRAINT [CO_Foo_Name_NOTNULL] NOT NULL,
[Description] NVARCHAR(max) NULL,
[LinkedBarId] int CONSTRAINT [CO_Foo_LinkedBarId_NOTNULL] NOT NULL,
[TypeId] int, -- CONSTRAINT [CO_Foo_TypeId_NOTNULL] NOT NULL,
CONSTRAINT [PK_Foo] PRIMARY KEY ([Id]),
CONSTRAINT [FK_Foo_TypeId]
FOREIGN KEY ([TypeId])
REFERENCES [dbo].[FooType] ([Id])
ON DELETE CASCADE
ON UPDATE CASCADE,
--CONSTRAINT [FK_Foo_LinkedBarId] -- named FOREIGN KEY constraint
--FOREIGN KEY ([LinkedBarId])
--REFERENCES [dbo].[Bar] ([Id])
-- ON DELETE CASCADE
-- ON UPDATE CASCADE,
)
GO
Separated from CREATE statement/batch, a constraint can be added in ALTER statement.
ALTER TABLE [dbo].[Bar]
ADD CONSTRAINT [PK_Bar_Id] PRIMARY KEY ([Id])
GO
ALTER TABLE [dbo].[Bar]
ADD CONSTRAINT [CO_Bar_Name_NOTNULL] -- named NOT NULL constraint
CHECK([Name] is NOT NULL)
GO
ALTER TABLE [dbo].[Foo]
ADD CONSTRAINT [DF_Foo_Description_NA] -- named DEFAULT constraint
DEFAULT ('N/A') FOR [Description]
GO
ALTER TABLE [dbo].[Foo]
ADD CONSTRAINT [CO_Foo_TypeId_NOTNULL] -- named NOT NULL constraint
CHECK([TypeId] is NOT NULL)
GO
ALTER TABLE [dbo].[Foo]
ADD CONSTRAINT [FK_Foo_LinkedBarId] -- named FOREIGN KEY constraint
FOREIGN KEY ([LinkedBarId])
REFERENCES [dbo].[Bar] ([Id])
ON DELETE CASCADE
ON UPDATE CASCADE
GO
The following query provides a view of all constraints in a database.
SELECT db_name() AS DbName, sys_table.name as TableName, user_name(sys_column.uid) as SchemaName, sys_column.name as ConstraintName, col.name as ColumnName, col.colid as OrdinalPosition, comments.text as DefaultClause FROM sysobjects sys_column JOIN syscomments comments ON sys_column.id = comments.id JOIN sysobjects sys_table ON sys_column.parent_obj = sys_table.id JOIN sysconstraints con ON sys_column.id = con.constid JOIN syscolumns col ON sys_table.id = col.id AND con.colid = col.colid WHERE sys_column.uid = user_id() AND sys_column.xtype = 'D' GO
See more at W3resource | Wiki.
Paged Collection in Data Service
It is naturally to use Expand in a Linq query to get children collection under an entity, as showed in the following example.
using System.Collections.Generic;
using System.Linq;
public IEnumerable<User> GetMembersByGroup(string groupIdentifier)
{
var team = this.dataContext
// each Team has a MemberUsers collection
.Teams.Expand(t => t.MemberUsers)
.Where(t => t.Id == groupIdentifier)
.Single();
return team.MemberUsers.ToList();
}
The above code would work as long as the size of the collection is small and within server paging size of the data service (- see this blog). In order to get result from all paged collection, the following example is using DataServiceCollection<T>.Load method.
Note: The service reference needs have UseDataServiceCollection enabled in .datasvcmap configuration which should be supported by .NET 3.5 SP1 and 4. In other case, rather than System.Data.Services.Client.DataServiceCollection, the System.Collections.ObjectModel.Collection won’t have Continuation.
using System.Collections.Generic;
using System.Data.Services.Client;
using System.Linq;
public IEnumerable<User> GetMembersByGroup(string groupIdentifier)
{
var team = this.dataContext
// expand both MemberUsers and Children teams
.Teams.Expand("MemberUsers,Children/MemberUsers")
.Where(t => t.Id == groupIdentifier)
.AsEnumerable()
.FirstOrDefault();
if (team == null)
{
return new List<User>();
}
DataServiceCollection<User> users = team.MemberUsers;
while (users.Continuation!= null)
{
users.Load(this.dataContext.Execute(users.Continuation));
}
return users;
}
The DataServiceCollection<T> requires a type T. For Query Projction with anonymous (or Tuple) type, the query can load data but may not support Continuation.
Another similar solution is sending data query to data service with GetContinuation.
using System.Collections.Generic;
using System.Linq;
public IEnumerable<User> GetMembersByGroup(string groupIdentifier)
{
var dataQuery =
from t in this.dataContext.Teams
where t.Id == groupIdentifier
from u in t.MemberUsers
select u;
var users = this.dataContext.GetAll<User>(dataQuery);
return users;
}
Since only navigation query supports join operation, the query must be on the primary key (as the Id in above code); otherwise, use another query with SingleOrDefault or FirstOrDefault in prior to get the identifier. Also, GetAll method should support Query Projection, so that GetAll(dataQuery) can be used, instead of GetAll((DataServiceQuery)dataQuery).
In order to query all users by a name (which is not a navigation query), the following data query projects the result to a collection of an anonymous typed objects, so that we can get identifiers for navigation queries later.
using System.Collections.Generic;
using System.Linq;
public IEnumerable<string> GetUsersByName(string userName)
{
var dataQuery =
from u in this.Users
where u.Name == userName
select new {
{
Id = u.Id, Alias = u.Alias
}
var result = this.dataContext.GetAll(dataQuery);
var users = result.Select(a => a.Id);
return users;
}
Here is the source of GetAll extension (with support of Query Projection):
using System.Collections.Generic;
using System.Data.Services.Client;
using System.Linq;
public static IEnumerable<T> GetAll<T>(
this DataServiceContext dataContext,
IQueryable<T> dataServiceQuery)
{
QueryOperationResponse<T> response =
(QueryOperationResponse<T>)
((DataServiceQuery<T>)dataServiceQuery).Execute();
DataServiceQueryContinuation<T> continuation = null;
do
{
if (continuation != null)
{
response = this.dataContext.Execute(continuation);
}
foreach (var result in response)
{
yield return result;
}
continuation = response.GetContinuation();
}
while (continuation != null);
}
