python代码片段

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运行结束

# 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()

>>> def fn(self, name='world'): # 先定义函数
...     print('Hello, %s.' % name)
...
>>> Hello = type('Hello', (object,), dict(hello=fn)) # 创建Hello class

# 导入:
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 创建对象的基类:
Base = declarative_base()

# 定义User对象:
class User(Base):
    # 表的名字:
    __tablename__ = 'user'

    # 表的结构:
    id = Column(String(20), primary_key=True)
    name = Column(String(20))

# 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://root:password@localhost:3306/test') # '数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
# 创建新User对象:
new_user = User(id='5', name='Bob')
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
session.commit()
# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()
# 关闭session:
session.close()

from wsgiref.simple_server import make_server

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    return '<h1>Hello, web!</h1>'

# 创建一个服务器，IP地址为空，端口是8000，处理函数是application:
httpd = make_server('', 8000, application)
print "Serving HTTP on port 8000..."
# 开始监听HTTP请求:
httpd.serve_forever()

app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def home():
    return '<h1>Home</h1>'

@app.route('/signin', methods=['GET'])
def signin_form():
    return '''<form action="/signin" method="post">
              <p><input name="username"></p>
              <p><input name="password" type="password"></p>
              <p><button type="submit">Sign In</button></p>
              </form>'''

@app.route('/signin', methods=['POST'])
def signin():
    # 需要从request对象读取表单内容：
    if request.form['username']=='admin' and request.form['password']=='password':
        return '<h3>Hello, admin!</h3>'
    return '<h3>Bad username or password.</h3>'

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('/opt/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.

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)

# 如下是一个装饰器，可以用在需要自动填充参数的函数上。功能是：
# 如果没有传递函数的deploy_id参数，那么就从环境变量中获取（调用自定义的get_global函数）
with_default_deploy_id = default_from_global('deploy_id', ENV_DEPLOYMENT)    

validator函数装饰func1，func1使用时接收参数(*arg, **kwargs)，而func1又装饰func2（其实就是Rally中的scenario函数），给func2增加validators属性，是一个函数的列表，函数的接收参数config, clients, task。这些函数最终调用func1，传入参数（config, clients, task, *args, **kwargs），所以func1定义时参数是（config, clients, task, *arg, **kwargs）  
最终实现的效果是，func2有很多装饰器，每个都会接收自己的参数，做一些校验工作。

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

class Foo(object):
    def __init__(self, *args, **kwargs):
        ...
    def __new__(cls, *args, **kwargs):
        return object.__new__(Stranger, *args, **kwargs)  

class Stranger(object):
    ...

foo = Foo()
print type(foo) 
# 打印的结果显示foo其实是Stranger类的实例。

class Singleton(object):  
    def __new__(cls, *args, **kw):  
        if not hasattr(cls, '_instance'):  
            cls._instance = super(Singleton, cls).__new__(cls, *args, **kw)  
        return cls._instance

class MyClass(Singleton): 
    pass

import functools

def singleton(cls):
    """Ensures single instance of a particular class."""
    instances = {}
    
    @functools.wraps(cls)
    def _singleton(*args, **kw):
        if cls not in instances:
            instances[cls] = cls(*args, **kw)
        return instances[cls]
    return _singleton

@singleton
class MyClass(BaseClass):
  pass

class Singleton(type):  
    def __init__(cls, name, bases, dict):  
        super(Singleton, cls).__init__(name, bases, dict)  
        cls._instance = None  
    def __call__(cls, *args, **kw):  
        if cls._instance is None:  
            cls._instance = super(Singleton, cls).__call__(*args, **kw)  
        return cls._instance  
  
class MyClass(object):  
    __metaclass__ = Singleton

import operator
def refresh_conn_infos(block_device_mapping, *refresh_args, **refresh_kwargs):
    map(operator.methodcaller('refresh_connection_info',
                              *refresh_args, **refresh_kwargs),
        block_device_mapping)
    return block_device_mapping

from datetime import datetime
def utc2local(utc_datetime):
    epoch = time.mktime(utc_datetime.timetuple())
    offset = datetime.fromtimestamp(epoch) - datetime.utcfromtimestamp(epoch)
    return utc_datetime + offset