分类 开发 下的文章

老坑,现在来填掉:)

之前做一个需求,是需要解析Flask里的URL Rule里的参数名字和类型,用来自动生成命令行内的Rest Client的参数。

Rule:  /disks/<int:disk_id>

需要得到参数: disk (type int)

解铃还须系铃人,直接看Flask源码吧:),看看它是如何管理/解析用户在 route 内添加的URL Parttern的.

首先,一路跟踪下去找到源码里的函数吧~~

from flask import Flask
 
app.route()

从 app.py中,可以看到

url_rule_class = Rule()

搜索parse 关键字,可以找到 parse_rule 函数,让我们一起来看看parse_rule函数吧:)

为了将parse_rule和 flask解耦,我拆出了以下代码。

import re

_rule_re = re.compile(r'''
    (?P<static>[^<]*)                           # static rule data
    <
    (?:
        (?P<converter>[a-zA-Z_][a-zA-Z0-9_]*)   # converter name
        (?:\((?P<args>.*?)\))?                  # converter arguments
        \:                                      # variable delimiter
    )?
    (?P<variable>[a-zA-Z_][a-zA-Z0-9_]*)        # variable name
    >
''', re.VERBOSE)


def parse_rule(rule):
    """Parse a rule and return it as generator. Each iteration yields tuples
    in the form ``(converter, arguments, variable)``. If the converter is
    `None` it's a static url part, otherwise it's a dynamic one.

    :internal:
    """
    pos = 0
    end = len(rule)
    do_match = _rule_re.match
    used_names = set()
    while pos < end:
        m = do_match(rule, pos)
        if m is None:
            break
        data = m.groupdict()
        if data['static']:
            yield None, None, data['static']
        variable = data['variable']
        converter = data['converter'] or 'default'
        if variable in used_names:
            raise ValueError('variable name %r used twice.' % variable)
        used_names.add(variable)
        yield converter, data['args'] or None, variable
        pos = m.end()
    if pos < end:
        remaining = rule[pos:]
        if '>' in remaining or '<' in remaining:
            raise ValueError('malformed url rule: %r' % rule)
        yield None, None, remaining

可以看到,flask讲一个URL Rule,首先拆解为 “/” 分隔的一个个单元。

之后,每个单元被解析为 converter(转换器), arguments(参数),variable(变量)。

以 /api/<int:id>/create 为例,可以知道得到三个单元

(None, None, "/api/"),
("int", None, "id"),
(None, None, "/create"),

第一个单元里,converter是None,则这个参数是一个URL中的静态字符串;

第二个单元内,converter是int,意即参数“id”是一个整型的变量;

第三个单元同第一个单元相同,不做多解释了。

通过以上分析,我们就拿到了Flask的Url Rule中变量的类型和名字,就可以做CLI自动生成中的参数解析了:)

PS:最近越来越赞同GongZi提的,“你会多少其实不重要要,重要的是你解决新问题的能力”,这种“Meta Learning”的能力,才是核心竞争力啊。

About:SQL数据库中,Index,字段类型等等直接影响整个SQL数据库的运行效率,有时候又必须考虑灵活性,所以才有这个话题:SQL数据库中的字段设计。

How:这里主要谈三个问题:

1.什么时候应该将一个属性作为单独的字段,什么时候可以打包。

2.index什么时候用

3.如何提高字段灵活性

- 阅读剩余部分 -

Why?

Web产品一旦上线,重启应用就会造成业务中断,对于实时性要求很高或者业务关联紧密的应用,重启程序是非常重的代价。

将代码对象序列化之后保存到存储内(比如redis, 关系数据库),在运行业务的时候通过制定的路由机制加载这部分业务。

对于线上应用,使用这种方式可以更加方便的部署新业务。

对于性能要求不是很高的场合,可以牺牲一些性能做代码动态装载。

- 阅读剩余部分 -

Python中的OS.fork,exec之类的行为,可能导致stdout所指向的FD发生变化,有时候无法在终端获得想要的输出,这时候可能需要在执行子进程的时候手工设置其FD。

参考:

http://stackoverflow.com/questions/26578799/python-send-sigint-to-subprocess-using-os-kill-as-if-pressing-ctrlc

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

简化并修改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,方便调试.当然也可以在运行时注册新的处理函数,对每个实例单独定制不同的转换函数(以适应不同的输出数据格式需求)

Class-Based view auto-init in init method.(Get safe GET paramters)

AttrDict with dict operation.

Validation by your own validator with wtforms.

Naming style in web development.