def itersubclasses(cls, _seen=None):
??? """Generator over all subclasses of a given class in depth first order."""
??? if not isinstance(cls, type):
??????? raise TypeError(_('itersubclasses must be called with '
????????????????????????? 'new-style classes, not %.100r') % cls)
??? _seen = _seen or set()
??? try:
??????? subs = cls.__subclasses__()
??? except TypeError:?? # fails only when cls is type
??????? subs = cls.__subclasses__(cls)
??? for sub in subs:
??????? if sub not in _seen:
??????????? _seen.add(sub)
??????????? yield sub
??????????? for sub in itersubclasses(sub, _seen):
??????????????? yield sub

import threading
is_done = threading.Event()
consumer = threading.Thread(
??? target=self.consume_results,
??? args=(key, self.task, runner.result_queue, is_done))
consumer.start()
self.duration = runner.run(
??????? name, kw.get("context", {}), kw.get("args", {}))
is_done.set()
consumer.join() #主線程堵塞，直到consumer運(yùn)行結(jié)束

The wait() method releases the lock, and then blocks until it is awakened by a notify() or notifyAll() call for the same condition variable in another thread. Once awakened, it re-acquires the lock and returns. It is also possible to specify a timeout.
The notify() method wakes up one of the threads waiting for the condition variable, if any are waiting. The notifyAll() method wakes up all threads waiting for the condition variable.
Note: the notify() and notifyAll() methods don't release the lock; this means that the thread or threads awakened will not return from their wait() call immediately, but only when the thread that called notify() or notifyAll() finally relinquishes ownership of the lock.

# Consume one item
cv.acquire()
while not an_item_is_available():
??? cv.wait()
get_an_available_item()
cv.release()
# Produce one item
cv.acquire()
make_an_item_available()
cv.notify()
cv.release()

class Timer(object):
??? def __enter__(self):
??????? self.error = None
??????? self.start = time.time()
??????? return self
??? def __exit__(self, type, value, tb):
??????? self.finish = time.time()
??????? if type:
??????????? self.error = (type, value, tb)
??? def duration(self):
??????? return self.finish - self.start
with Timer() as timer:
??? func()
return timer.duration()

class ModelMetaclass(type):
??? def __new__(cls, name, bases, attrs):
??????? if name=='Model':
??????????? return type.__new__(cls, name, bases, attrs)
??????? mappings = dict()
??????? for k, v in attrs.iteritems():
??????????? if isinstance(v, Field):
??????????????? print('Found mapping: %s==>%s' % (k, v))
??????????????? mappings[k] = v
??????? for k in mappings.iterkeys():
??????????? attrs.pop(k)
??????? attrs['__table__'] = name # 假設(shè)表名和類(lèi)名一致
??????? attrs['__mappings__'] = mappings # 保存屬性和列的映射關(guān)系
??????? return type.__new__(cls, name, bases, attrs)
class Model(dict):
??? __metaclass__ = ModelMetaclass
??? def __init__(self, **kw):
??????? super(Model, self).__init__(**kw)
??? def __getattr__(self, key):
??????? try:
??????????? return self[key]
??????? except KeyError:
??????????? raise AttributeError(r"'Model' object has no attribute '%s'" % key)
??? def __setattr__(self, key, value):
??????? self[key] = value
??? def save(self):
??????? fields = []
??????? params = []
??????? args = []
??????? for k, v in self.__mappings__.iteritems():
??????????? fields.append(v.name)
??????????? params.append('?')
??????????? args.append(getattr(self, k, None))
??????? sql = 'insert into %s (%s) values (%s)' % (self.__table__, ','.join(fields), ','.join(params))
??????? print('SQL: %s' % sql)
??????? print('ARGS: %s' % str(args))
class Field(object):
??? def __init__(self, name, column_type):
??????? self.name = name
??????? self.column_type = column_type
??? def __str__(self):
??????? return '<%s:%s>' % (self.__class__.__name__, self.name)
class StringField(Field):
??? def __init__(self, name):
??????? super(StringField, self).__init__(name, 'varchar(100)')
class IntegerField(Field):
??? def __init__(self, name):
??????? super(IntegerField, self).__init__(name, 'bigint')
class User(Model):
??? # 定義類(lèi)的屬性到列的映射：
??? id = IntegerField('id')
??? name = StringField('username')
??? email = StringField('email')
??? password = StringField('password')
# 創(chuàng)建一個(gè)實(shí)例：
u = User(id=12345, name='Michael', email='test@orm.org', password='my-pwd')
# 保存到數(shù)據(jù)庫(kù)：
u.save()

Found model: User
Found mapping: email ==>
Found mapping: password ==>
Found mapping: id ==>
Found mapping: name ==>
SQL: insert into User (password,email,username,uid) values (?,?,?,?)
ARGS: ['my-pwd', 'test@orm.org', 'Michael', 12345]

# 導(dǎo)入:
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
# 創(chuàng)建對(duì)象的基類(lèi):
Base = declarative_base()
# 定義User對(duì)象:
class User(Base):
??? # 表的名字:
??? __tablename__ = 'user'
??? # 表的結(jié)構(gòu):
??? id = Column(String(20), primary_key=True)
??? name = Column(String(20))
# 初始化數(shù)據(jù)庫(kù)連接:
engine = create_engine('mysql+mysqlconnector://root:password@localhost:3306/test') # '數(shù)據(jù)庫(kù)類(lèi)型+數(shù)據(jù)庫(kù)驅(qū)動(dòng)名稱://用戶名:口令@機(jī)器地址:端口號(hào)/數(shù)據(jù)庫(kù)名'
# 創(chuàng)建DBSession類(lèi)型:
DBSession = sessionmaker(bind=engine)
# 創(chuàng)建新User對(duì)象:
new_user = User(id='5', name='Bob')
# 添加到session:
session.add(new_user)
# 提交即保存到數(shù)據(jù)庫(kù):
session.commit()
# 創(chuàng)建Query查詢，filter是where條件，最后調(diào)用one()返回唯一行，如果調(diào)用all()則返回所有行:
user = session.query(User).filter(User.id=='5').one()
# 關(guān)閉session:
session.close()

from wsgiref.simple_server import make_server
def application(environ, start_response):
??? start_response('200 OK', [('Content-Type', 'text/html')])
??? return '

Hello, web!

'
# 創(chuàng)建一個(gè)服務(wù)器，IP地址為空，端口是8000，處理函數(shù)是application:
httpd = make_server('', 8000, application)
print "Serving HTTP on port 8000..."
# 開(kāi)始監(jiān)聽(tīng)HTTP請(qǐng)求:
httpd.serve_forever()

from flask import Flask
from flask import request
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def home():
??? return '

Home

'
@app.route('/signin', methods=['GET'])
def signin_form():
??? return '''

?????????????

?????????????

?????????????

Sign In

?????????????

'''
@app.route('/signin', methods=['POST'])
def signin():
??? # 需要從request對(duì)象讀取表單內(nèi)容：
??? if request.form['username']=='admin' and request.form['password']=='password':
??????? return '

Hello, admin!

'
??? return '

Bad username or password.

'
if __name__ == '__main__':
??? app.run()

print(json.dumps(data, indent=4))
# 或者
import pprint
pprint.pprint(data)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import itertools
import sys
class ImmutableMixin(object):
??? _inited = False
??? def __init__(self):
??????? self._inited = True
??? def __setattr__(self, key, value):
??????? if self._inited:
??????????? raise Exception("unsupported action")
??????? super(ImmutableMixin, self).__setattr__(key, value)
class EnumMixin(object):
??? def __iter__(self):
??????? for k, v in itertools.imap(lambda x: (x, getattr(self, x)), dir(self)):
??????????? if not k.startswith('_'):
??????????????? yield v
class _RunnerType(ImmutableMixin, EnumMixin):
??? SERIAL = "serial"
??? CONSTANT = "constant"
??? CONSTANT_FOR_DURATION = "constant_for_duration"
??? RPS = "rps"
if __name__=="__main__":
??? print _RunnerType.CONSTANT

import os
def write_to_file(path, contents, umask=None):
??? """Write the given contents to a file
??? :param path: Destination file
??? :param contents: Desired contents of the file
??? :param umask: Umask to set when creating this file (will be reset)
??? """
??? if umask:
??????? saved_umask = os.umask(umask)
??? try:
??????? with open(path, 'w') as f:
??????????? f.write(contents)
??? finally:
??????? if umask:
??????????? os.umask(saved_umask)
if __name__ == '__main__':
??? write_to_file('/home/kong/tmp', 'test', 31)
??? # Then you will see a file is created with permission 640.
??? # Warning: If the file already exists, its permission will not be changed.
??? # Note：For file, default all permission is 666, and 777 for directory.

import multiprocessing
import time
import os
def run(flag):
??? print "flag: %s, sleep 2s in run" % flag
??? time.sleep(2)
??? print "%s exist" % flag
??? return flag
if __name__ == '__main__':
??? pool = multiprocessing.Pool(3)
??? iter_result = pool.imap(run, xrange(6))
??? print "sleep 5s\n\n"
??? time.sleep(5)
??? for i in range(6):
??????? try:
??????????? result = iter_result.next(600)
??????? except multiprocessing.TimeoutError as e:
??????????? raise
??????? print result
??? pool.close()
??? pool.join()

import decorator
def default_from_global(arg_name, env_name):
??? def default_from_global(f, *args, **kwargs):
??????? id_arg_index = f.func_code.co_varnames.index(arg_name)
??????? args = list(args)
??????? if args[id_arg_index] is None:
??????????? args[id_arg_index] = get_global(env_name)
??????????? if not args[id_arg_index]:
??????????????? print("Missing argument: --%(arg_name)s" % {"arg_name": arg_name})
??????????????? return(1)
??????? return f(*args, **kwargs)
??? return decorator.decorator(default_from_global)
# 如下是一個(gè)裝飾器，可以用在需要自動(dòng)填充參數(shù)的函數(shù)上。功能是：
# 如果沒(méi)有傳遞函數(shù)的deploy_id參數(shù)，那么就從環(huán)境變量中獲取（調(diào)用自定義的get_global函數(shù)）
with_default_deploy_id = default_from_global('deploy_id', ENV_DEPLOYMENT)???

def validator(fn):
??? """Decorator that constructs a scenario validator from given function.
??? Decorated function should return ValidationResult on error.
??? :param fn: function that performs validation
??? :returns: rally scenario validator
??? """
??? def wrap_given(*args, **kwargs):
??????? """Dynamic validation decorator for scenario.
??????? :param args: the arguments of the decorator of the benchmark scenario
??????? ex. @my_decorator("arg1"), then args = ('arg1',)
??????? :param kwargs: the keyword arguments of the decorator of the scenario
??????? ex. @my_decorator(kwarg1="kwarg1"), then kwargs = {"kwarg1": "kwarg1"}
??????? """
??????? def wrap_validator(config, clients, task):
??????????? return (fn(config, clients, task, *args, **kwargs) or
??????????????????? ValidationResult())
??????? def wrap_scenario(scenario):
??????????? wrap_validator.permission = getattr(fn, "permission",
??????????????????????????????????????????????? consts.EndpointPermission.USER)
??????????? if not hasattr(scenario, "validators"):
??????????????? scenario.validators = []
??????????? scenario.validators.append(wrap_validator)
??????????? return scenario
??????? return wrap_scenario
??? return wrap_given