酒吞一馆

好友新婚。酒吞一贯。小聚。上海滩最好的日料。

因为离公司近,所以有些仪式感的事情,会去那里。

2013年10月,人生最低谷的时候,在那里,可惜即便如此美味,也咽不下去。

所以,美食,还是要有心情搭配。

迁移“创意纪”的 blog 到这里

创意纪这个 blog,域名用了很多年,写了很多年。中间有一次从 mediatemple 迁移的时候,没有做好,2013年从头开始写了一些。陪伴我度过了生命中一段艰难的岁月。

不想维护太多blog,明日会目前使用了一段时间,还是国内的访问速度稳定。

准备用 wordpress 的 export 和 import 功能迁移一下数据,至少文字都能记录下来就好。

python flask 写 api 如何返回自定义错误

在 python 开发中,利用 flask 写 restful api 函数的时候,除了标准的400、500等这些返回码通过 abort() 返回以外,怎么另外返回自定义的错误代码和信息呢?

我们碰到的业务场景是对于api 输入参数的各类校验以及在业务逻辑执行的时候,都会返回统一的400代码,同时也会返回我们约定的描述详细错误的代码以及描述字符串,提供给调用方来处理,这样可以让其用户体验做得更好,同时详细错误代码和描述字符串也会自动打印在 log 日志中。

flask 的官方文档中告诉我们:

默认情况下,错误代码会显示一个黑白的错误页面。如果你要定制错误页面, 可以使用 errorhandler() 装饰器

在写 restful api 的时候,并没有页面可以返回,我们可以在 flask 提供的代码基础上稍加改造如下。

在你的初始化 flask app 的相关代码中加入下面两个函数:

@app.errorhandler(CustomFlaskErr)
def handle_flask_error(error):

    # response 的 json 内容为自定义错误代码和错误信息
    response = jsonify(error.to_dict())

    # response 返回 error 发生时定义的标准错误代码
    response.status_code = error.status_code

    return response
class CustomFlaskErr(Exception):

    # 默认的返回码
    status_code = 400

    # 自己定义了一个 return_code,作为更细颗粒度的错误代码
    def __init__(self, return_code=None, status_code=None, payload=None):
        Exception.__init__(self)
        self.return_code = return_code
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    # 构造要返回的错误代码和错误信息的 dict
    def to_dict(self):
        rv = dict(self.payload or ())

        # 增加 dict key: return code
        rv['return_code'] = self.return_code

        # 增加 dict key: message, 具体内容由常量定义文件中通过 return_code 转化而来
        rv['message'] = J_MSG[self.return_code]

        # 日志打印
        logger.warning(J_MSG[self.return_code])

        return rv

CustomFlaskErr 是我们自己写的处理错误的类,然后通过 @app.errorhandler(CustomFlaskErr) 这个装饰器在 flask 中注册。

具体功能在注释里基本都写了,我们看一下怎么使用这个自定义错误处理器。

# 用户名输入为空
if user_name is None:
    raise CustomFlaskErr(USER_NAME_ILLEGAL, status_code=400)

当需要处理某个错误的时候,rasie 刚才的 CustomFlaskErr,传递另外定义好的自己的错误代码,以及标准的返回代码;

上面说的常量定义文件可以参考如下:

USER_ALREADY_EXISTS = 20001  # 用户已经存在
J_MSG = {USER_ALREADY_EXISTS: 'user already exists'}

通过这样的机制,就做到了在具体 restful api 的业务逻辑代码中简单的进行各类自定义错误的处理,所有的错误处理是集中的,细颗粒度的错误代码和消息也是集中维护,便于扩展。

flask 官方文档和一些网上的资料都说比较简单,实践中摸索了这样的实现方式供参考。

python 中使用装饰器来统一检查 flask 用户权限

最近在一个项目中,需要判断 restful 接口函数传入的时候,是否之前已经登录状态是某个特定用户,以及该用户有没有指定的权限。检查下来如果没有的话,立刻返回错误,中断功能。

遮掩的场景虽然也可以通过标准的调用函数来操作,但都不如用装饰器来得简单。都知道装饰器好用不好写,废话不说,先来看看这个场景怎么实现,还是有一定的通用性的。

def validate_current_is_admin(f):
    @functools.wraps(f)
    def decorated_function(*args, **kws):
        # 需要在登录状态调用, 检查是否为有admin权限的用户登录,
        # 如果不是,返回错误码;
        if g.user.user_name != 'admin':
            raise CustomFlaskErr(USER_MUST_HAS_ADMIN_PRIVILEGE, status_code=401)

        # 验证权限是否为 admin, 不是的话,返回401错误
        if g.user.role_id != Permission.ADMIN:
            raise CustomFlaskErr(USER_MUST_HAS_ADMIN_PRIVILEGE, status_code=401)

        return f(*args, **kws)

    return decorated_function

这是一个标准的装饰器的写法,如果你要写一个简单的装饰器,整个框架可以参考。

装饰器调用举例:

@app.route('/api/create_user', methods=['POST'])
@auth.login_required
@validate_current_is_admin
def create_user():

    # 获得参数
    user_name = request.json.get('user_name')
    password = request.json.get('password')
......

 

核心代码的业务逻辑也不复杂,根据 flask 的 g 对象中预存的用户 user 进行检查处理,flask 的这些定义非常灵活,flask.g 怎么使用可以查看 flask 的文档。

这里的 user 以及相关的属性属于具体业务逻辑,就不展开解释了,可以望文生义。

因为不对 args 和 kws 这些参数进行解析和处理,所处理的是 flask 全局对象。最后将参数都原路打包返回即可,没有问题的话交给使用装饰器的代码继续处理。

这个例子比较简单,主要还是熟悉装饰器的基本用法。

联想为什么

隐约记得以前有本书叫做:联想为什么,说的是联想发家的故事。

记得很久以前,一个同事买联想的家用电脑,我还特地去帮忙检查什么的。在1996年的时候,买一个这样电脑是非常令人值得开心的事情。

去年,还帮家人买过一个联想的笔记本,其实我是不喜欢的,造型也不是很漂亮,并且价格很贵,家人比较执拗,也就算了。

看到联想2017年最新季报净亏损7200万美元,而 CEO 杨元庆居然年薪 1924万美元。

2013-2014年的时候,联想的手机还是前五名吧,现在完全没有任何人使用。台式机我也看不到有人使用,当然现在买台式机的人本来就少了。

我周围的同事朋友的笔记本基本都是 apple、dell、hp,或者游戏本。

还是温水煮青蛙,有时候,就这样不行了,好像没有错,就是错了。实际上,曾经得到太容易,没有危机感。

联想的管理体制现在看来是有很大的问题,曾经的民族骄傲,可以买下 thinkpad,不知道这样还能撑多久。

python 之前因后果

如果你英语阅读还可以的话,推荐这个网站:The History of Python

2013年11月后就不再更新了,一共31篇 blog,讲的是 python 语言设计中的一些来龙去脉。我在搜索研究列表生成式的时候偶然看到了这个网站。

挺有趣的。

它更新的最后一篇文章是 The history of bool, True and False

或许你也像我一样被 python 的布尔值稍稍困惑过,我们可以看看当年这些天才的程序语言设计者到底是怎么想的。

2017.8.21 由微信部分功能故障所想

像微信这样的程序以及背后的架构,是非常复杂,也非常牛 x 的,春节这么发红包都不会弄垮它。可惜,只要是人写的系统,总是会有意外发生。下午,有同事说微信不能转账了,后来看到新闻说下午微信部分功能不能使用,包括支付、公众号、分享等。下午四点十分左右,据说所有功能恢复。

作为信息中心的一员,每次遇到故障总是最令人揪心的,这也是最近几个月工作强度非常大的原因之一,为了让系统稳定健壮。

现在想起来,技术发展的空间的确还很大,服务器、数据库等技术这些年已经经历了几百倍甚至更多的性能的增加,而系统的复杂度也是大大增加了,记得以前 CS 架构、三层架构,基本就可以解决问题了。现在我们研究的是 micro service、serviceless、灰度、内存数据库、流式计算等等。有趣而烧脑。

python 单元测试中使用参数化测试技巧(parameterisation)

这篇文章介绍了在 python 的单元测试中如何使用参数化测试(parameterisation) 技巧来做到将测试数据和测试逻辑分离。Improve Python testing with parameterisation

的确,我们在应用程序中对于业务逻辑和数据的分离会做很好的考虑,但是在自动化测试,在单元测试中有时候会忽略这一点,我发现我也有这个问题,所以写出来的单元测试类似这样:

# test_prime.py
import unittest

from prime import is_prime


class TestIsPrime(unittest.TestCase):

    def test_x_negative(self):
        self.assertEqual(is_prime(-1), False)

    def test_x_zero(self):
        self.assertEqual(is_prime(0), False)

    def test_x_one(self):
        self.assertEqual(is_prime(1), False)

    def test_x_two(self):
        self.assertEqual(is_prime(2), True)

    def test_x_three(self):
        self.assertEqual(is_prime(3), True)

    def test_x_ten(self):
        self.assertEqual(is_prime(10), False)

    def test_x_fifty_three(self):
        self.assertEqual(is_prime(53), True)


if __name__ == "__main__":
    unittest.main()

而这篇文章中建议我们应该这样:

import unittest

from prime import is_prime


class TestIsPrime(unittest.TestCase):

    def test_is_prime(self):
        test_cases = [
            (-1, False),
            (0, False),
            (1, False),
            (2, True),
            (3, True),
            (10, True),
            (53, True),
        ]
        for x, output in test_cases:
            with self.subTest(name=str(x)):
                self.assertEqual(is_prime(x), output)


if __name__ == "__main__":
    unittest.main()

好处不言而喻。

python 3.x 的线程池模式实现多线程

之前发布在简书,我觉得简书还是一个做得非常好的写作平台,也有一定的互动,我现在比较纠结的就是在受众的广度、受众的亲疏、自己可以掌握的自由度、维护成本这些变量之间。

比如问题之一,简书虽然有很好的写作平台,但是我就是喜欢在 wordpress 里面自己折腾主题等等,怎么办。哎。

先从简书这里搬一些自己的文章过来。


看了不少书和资料,自认为对于 python 中的线程、进程、协程等略知一二了。

想实现一个多线程池的模型,但是也不想用 queue 甚至是 celery 这些,查了很多资料,国内的原创的不多,并且基本都是停留在 python 2.7 的时代,而且国内的文章即便用 google 搜索,大部分文章也是互相转载。国外的资料比较好的还是在 stackoverflow,国内的简书上好文章不少。

搜了半天,在 python 的官方文档上赫然有着一个例子:

The concurrent.futures module provides a high-level interface for asynchronously executing callables.
The asynchronous execution can be performed with threads, using ThreadPoolExecutor, or separate processes, using ProcessPoolExecutor. Both implement the same interface, which is defined by the abstract Executor class.

用 concurrent.futures 即可,然后

ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.

找了好久差点想自己实现(担心自己写的很烂)的线程池就在这里,例子如下:

import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

这个例子写得很清楚,可以直接运行,不过建议把那些网址换掉,因为 g*w 的关系。

没想到 python 的官方文档做的这么好,我准备从头到底先通读一遍。

从学习角度这个代码是够了,当然如果要用在真正的应用里面,还要考虑更多的事情哦!

银行还能这么继续想?

偶尔听一位朋友转述,某位银行管理层认为的几个观点:

1 互联网金融不是那么可靠的,绝对不能把钱放在这样的企业里面
2 互联网金融,包括支付宝这些,安全肯定不怎么样,银行账户之所以安全是因为它是封闭的,而互联网这些账户直接暴露在网络上,黑客很容易攻击,然后就什么都没有了
3 央行早晚会出规定,大额的都不允许走第三方支付公司

其实这几年国家政策已经非常倾向传统银行了,为了保护银行利益,对于第三方支付设置了一二三级账户等等。还有最近余额宝的额度已经被限制到10万了,从100万到50万到10万……

互联网金融中,有一些打着 p2p 或者种种旗号,欺骗老百姓,卷款而逃的,的确有那么一批。我觉得互联网金融短短几年,蓬勃发展,这些是阵痛。
但是,传统银行门一开,钱就来的好日子恐怕一去不复返了。

科技、效率、用户体验、创新等,其实银行某种程度上,已经因为支付宝、微信支付、其他第三方支付、大量的互联网金融创新倒逼之下,必须接受新的游戏规则。

期待央行出一些严苛的规定,不是不可以,只是不要在等这些规定救命的时候已经挂了。

我眼中的银行现在只有两种,一种是锐意进取,拥抱创新,另外一种有点像开头说的那个,依然夜郎自大,被温水煮青蛙,没有意识到危机已经来临。

任何因为故步自封而倒下的企业,都是轰然一声。太多企业家说过了,居安思危,不管你认为企业在什么时期。

记录一下,看看中国的第三方支付和互联网金融到底怎么发展,是不是所有的银行还能享受政策红利,那么容易的赚钱。