用动态加载和代码动态生成提高应用灵活性
Why？
Web产品一旦上线，重启应用就会造成业务中断，对于实时性要求很高或者业务关联紧密的应用，重启程序是非常重的代价。
将代码对象序列化之后保存到存储内（比如redis， 关系数据库），在运行业务的时候通过制定的路由机制加载这部分业务。
对于线上应用，使用这种方式可以更加方便的部署新业务。
对于性能要求不是很高的场合，可以牺牲一些性能做代码动态装载。
2017年8月
OS.fork等子进程调用中FD的变化
Python中的OS.fork，exec之类的行为，可能导致stdout所指向的FD发生变化，有时候无法在终端获得想要的输出，这时候可能需要在执行子进程的时候手工设置其FD。
参考：
http://ipseek.blog.51cto.com/1041109/809467
http://stackoverflow.com/questions/9211190/python-fork-pipe-and-exec
http://bytes.com/topic/python/answers/836323-problem-threading-execv
Pyramid中简化你的代码
简化并修改view_config装饰器
因为Pyramid自带的view_config装饰器非常复杂，需要配置的参数比较多，修改一下装饰器，每次可以少写一点代码……
def get_with_permission(**settings):
# delete debug tag TODO
view_configer = view_config(decorator=permission_require(login_url=LOGIN_URL), **settings)
return view_configer
def p_get(route_name, renderer='json', **kwargs):
return get_with_permission(route_name=route_name, renderer=renderer, request_method="GET", **kwargs)
这样就可以将复杂的装饰器包裹在这个新装饰器内部，也可以简单的定制，只需要输入route_name就可以
login_required decorator and more
def permission_require(view_callable=None, login_url=None):
"""
Return a login_required decorator with login_url redirection.
Usage: @view_config(decorator=login_required, **configs) or @view_config(decorator=login_required(login_url='/'), **configs)
:param view_callable: pyramid view_callable
:param login_url: login url setting
:type login_url: str or unicode
:return: callable
"""
if not view_callable and login_url:
if not isinstance(login_url, (str, unicode)):
raise TypeError('login_url must be a str/unicode object!')
return lambda func: permission_check_actual(func, login_url)
else:
return permission_check_actual(view_callable, '/accounts/login')
以上是使用Pyramid自带的decorator特性定制的LoginRequired装饰器。
这里能获取到的函数参数和自写的装饰器是不同的，因为这里你可以取到request对象和context对象，但位于view_config外部的装饰器则很麻烦。
和外部装饰器搭配使用，可以获得更好的效果，可以在装饰器内部做登录和针对request对象或者request对象包含的任意成员的鉴权
SqlAlchemy 类方法。
class OperaionLog(Base):
__tablename__ = 'operation_log'
id = Column(Integer, primary_key=True, autoincrement=True)
uid = Column(Integer, nullable=False)
username = Column(Text)
operation = Column(Text, nullable=False, doc=u'操作内容')
result = Column(Text, doc=u'操作结果')
ctime = Column(DateTime, nullable=False, default=datetime.datetime.utcnow)
@classmethod
def log_me(cls, uid, username, operation, result):
with transaction.manager:
log = cls(
uid=uid,
username=username,
operation=operation,
result=result,
)
dbsession.add(log)
dbsession.flush()
return True
@classmethod
def get_many(cls, start, end, is_dict=True):
if is_dict:
dbsession.query(cls).order_by(desc(cls.ctime))
以上代码描述了一个SQLAlchemy类方法，将对ORM的操作全部封装在ORM类当中，可以极大程度的精简查询代码，不用每个视图都写一次，且可以保证多处的查询获得一致的结果（比如开发中对于数据库字段状态的判断，不同开发时期很容易造成查询条件不一致）
自动数据库字段转换
虽然有很多项现成的Alchenmy2form这种类型的库，但对于输出数据或者对于要求输出JSON数据的情况下，使用REST-FRAMEWORK会比较重，所以使用一个自动格式化的工具会极大较少工作量又不用使用巨大的框架
class DBFieldConverter(object):
"""
Convert specified SQLAlchemy model to output dict or other format.
"""
_registry = {}
def __init__(self, model_instance, allow_output=[], registry={}, extra_out=[]):
"""
If registry is given, same method in this registry will overwrite that in class _registry.
:param model_instance:
SQLAlchemy model instance, inherit from declarative_base().
:type registry: dict
:type allow_output: list or tuple
"""
if not hasattr(model_instance, '__table__'):
raise TypeError("model_instance must be instance of sqlalchemy's model")
if not isinstance(allow_output, (list, tuple)):
raise TypeError("allowed outputs [{allow_output}] must be list or tuple")
if not isinstance(registry, dict):
raise TypeError('argument registry `{registry}` must be a dict.'.format(registry=registry))
self.registry = registry
self.model = model_instance
self._allows = allow_output
self.dict = dict((col, self._convert(getattr(self.model, col), type(self.model.__table__.columns[col].type))) for col in self.model.__table__.columns.keys())
for key in extra_out:
self.dict[key] = getattr(model_instance, key)
def as_dict(self, pure=False):
if pure:
return self.dict
else:
return dict((item for item in self.dict.items() if item[0] in self._allows))
def as_list(self, pure=False):
if pure:
return self.dict.items()
else:
return (item for item in self.dict.items() if item[0] in self._allows)
@classmethod
def register(cls, data_type, convert_method=None):
"""
Register a field type converter to FieldConverter.
Pass convert_method param or use default convert_method.
If the data_type existed, this will return default
:type data_type: wtforms.fields.Field
:type convert_method: callable
:rtype :bool
:return True if register successfully, else False.
"""
if not isinstance(data_type, type):
raise TypeError('{data_type} must be instance of type'.format(data_type=data_type))
if convert_method is not None:
if not callable(convert_method):
raise TypeError('{convert_method} must be callable'.format(convert_method=convert_method))
else:
convert_method = lambda field: field
cls._registry[data_type] = convert_method
return True
@classmethod
def unregister(cls, data_type):
"""
Unregister a data-type converter fucntion.
:type data_type: type
:rtype bool
:return True if register successfully
"""
if not isinstance(data_type, type):
raise TypeError('{data_type} must be instance of type'.format(data_type=data_type))
if cls._registry.get(data_type, None) is None:
logging.warning("Type {data_type} does not exist in registry.".format(data_type=data_type))
else:
del cls._registry[data_type]
return True
def _convert(self, data, data_type):
if data is None:
return None
if self.registry.get(data_type) is not None:
return self.registry[data_type](data)
elif self._registry.get(data_type) is not None:
return self._registry[data_type](data)
else:
logging.warning('{type} not contained in registry, return its default value'.format(type=data_type))
return data
DBFieldConverter.register(ARRAY, _list2unicode) #自定义的转换函数
DBFieldConverter.register(JSON, json.dumps)
DBFieldConverter.register(Text)
DBFieldConverter.register(String)
DBFieldConverter.register(DateTime, datetime2utmp) #自定义的转换函数
DBFieldConverter.register(Integer)
DBFieldConverter.register(CHAR)
DBFieldConverter.register(Boolean)
DBFieldConverter.register(UUID)
DBFieldConverter.register(INET)
以上代码使用白名单机制输出数据，指定输出的字段名即可，也可以使用pure=True，方便调试.当然也可以在运行时注册新的处理函数，对每个实例单独定制不同的转换函数（以适应不同的输出数据格式需求）