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

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 # 假设表名和类名一致
        attrs['__mappings__'] = mappings # 保存属性和列的映射关系
        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):
    # 定义类的属性到列的映射：
    id = IntegerField('id')
    name = StringField('username')
    email = StringField('email')
    password = StringField('password')
# 创建一个实例：
u = User(id=12345, name='Michael', email='test@orm.org', password='my-pwd')
# 保存到数据库：
u.save()

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

# 导入:
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()

from flask import Flask
from flask import request
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('/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)
# 如下是一个装饰器，可以用在需要自动填充参数的函数上。功能是：
# 如果没有传递函数的deploy_id参数，那么就从环境变量中获取（调用自定义的get_global函数）
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