クラスのすべてのサブクラスを取得

 
  
 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     
  
 
 

もっと話してEvent()はthreadingに置き換えることもできる.Condition()、conditionにはnotify()、wait()、notifyAll()があります.次のように解釈されます.

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

メタクラス
__new__()メソッドが受信するパラメータは、現在作成するクラスのオブジェクトの順です.クラスの名前;クラスが継承する親クラスの集合;クラスのメソッドの集合;

 
  
 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 '' % (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='[email protected]', password='my-pwd') 
  
 #       ： 
  
 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', '[email protected]', 'Michael', 12345] 
  
 
 

SQLAlchemy簡単に使用

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

WSDIのシンプルな使用とWebフレームワークFlaskのシンプルな使用

 
  
 from wsgiref.simple_server import make_server 
  
 def application(environ, start_response): 
  
     start_response('200 OK', [('Content-Type', 'text/html')]) 
  
     return '
Hello, web!
' 
  
 #        ，IP    ，   8000，     application: 
  
 httpd = make_server('', 8000, application) 
  
 print "Serving HTTP on port 8000..." 
  
 #     HTTP  : 
  
 httpd.serve_forever() 
  
 
 

WSGIフレームワークを理解して、私達は発見します:実は1つのWeb App、1つのWSGIの処理関数を書いて、各HTTPの要求に対して応答します.しかし、HTTPリクエストをどのように処理するかは問題ではなく、100の異なるURLをどのように処理するかが問題です.最も単純で最も地味な考え方はenviron変数からHTTPリクエストの情報を取り出し,一つ一つ判断することである.

 
  
 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        ： 
  
     if request.form['username']=='admin' and request.form['password']=='password': 
  
         return '
Hello, admin!
' 
  
     return '
Bad username or password.
' 
  
 if __name__ == '__main__': 
  
     app.run() 
  
 
 

フォーマット表示json

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

JavaやCのような列挙を実現

 
  
 #!/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

" 
  
     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関数はfunc 1を装飾し、func 1は使用時にパラメータ(*arg、**kwargs)を受信し、func 1はfunc 2(実はRallyのscenario関数)を装飾し、func 2にvalidators属性を追加し、関数のリストであり、関数の受信パラメータconfig、clients、taskである.これらの関数は最終的にfunc 1を呼び出し、パラメータ(config,clients,task,*args,*kwargs)を入力するので、func 1定義時のパラメータは(config,clients,task,*arg,*kwargs)最終的に実現される効果は、func 2には多くの装飾器があり、それぞれが自分のパラメータを受信し、いくつかの検査作業を行うことです.

 
  
 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 
  
 
 

inspectライブラリの一般的な使い方
inspect.getargspec(func)は関数パラメータの名前とデフォルト値を取得し、args(args,varargs,keywords,defaults)を返します.argsはパラメータ名のリストです.varargsとkeywordsは*番と**番の変数名です.defaultsはパラメータのデフォルト値のリストです.
inspect.getcallargs(func[,* args][,**kwds])は関数パラメータをバインドします.バインドされた関数のパラメータ辞書を返します.
pythonのプライベート属性と関数
Pythonは、2つ以上の下線文字で始まり、2つ以上の下線で終わる変数がない変数をプライベート変数とします.プライベート変数は、クラスAの__のように、コード生成前に長いフォーマット(公有化)に変換されます.このプロセスは「Private name mangling」と呼ばれます.private識別子は_に変換されます.A__privateですが、クラス名がすべて以下の線で命名されると、Pythonは圧延を実行しません.また、プライベート変数と呼ばれていますが、アクセスまたは変更される可能性があります(_classname__membernameを使用)ので、以下にまとめます.
単一下線でも二重下線でも先頭のメンバーは、外部プログラム開発者がこれらのメンバー変数とこれらのメンバー関数を直接使用しないことを望んでいます.ただし、二重下線は文法的に誤りを避けることができますが、_クラス名_メンバー名は依然としてアクセスできます.単一の下線は、ダイナミックデバッグ時に便利ですが、プロジェクトグループの人が下線の先頭のメンバーを守って直接使用しない限り、単一の下線を使用したほうがいいかもしれません.