這篇文章摘自我的博客, 歡迎大家沒事去逛逛~

這幾個月我開發了公司里的一個restful webservice，起初技術選型的時候是采用了flask框架。雖然flask是一個同步的框架，但是可以配合gevent或者其它方式運行在異步的容器中(測試鏈接)，效果看上去也還可以，因此就采用了這種方式。

后面閱讀了tornado的源碼，也去了解了各種協程框架以及運行的原理。總感覺flask的這種同步方式編程不夠好，同時對于這種運行在容器里的模式目前還缺乏了解。但至少現在對于tornado的運行原理有了一定的了解，如果用tornado寫的話，是很可控的，而且可以保證運行是高效的。因此就決定把原來基于flask的項目用tornado重構了。

項目重構的過程中遇到了一些問題，也學習了一些東西，這里做一個簡單的總結。

所有框架都要處理的一個接入層的事情就是：

url-mapping

項目初始化

參數解析

對于restful風格的接口以及項目的初始化，每個框架都有自己的方式，在它們的文檔中都演示得特別清楚，所以關于這些我就不展開了。

關于參數解析，這里并不是指簡單地調用類似于get_argument這樣的方法去獲取數據。而是 如何從不可靠的client端傳來的數據中過濾掉服務器不關注的數據，同時對服務器關注的數據作一些更強的校驗，這就是協議層的事情了。

使用谷歌的ProtocolBuffer是一個不錯的方案，它有很不錯的數據壓縮率，也支持目前大多數主流的開發語言。但在一些小型項目中，我還是更偏向于使用json的方式，它顯得更加靈活。但是對于json的話，如何作數據校驗就是另外一個問題了。

在重構前，我是通過python中的裝飾器來實現的這個功能:

class SomeHandlerInFlask(Resource):
    @util.deco({
        "key_x": (str, "form"),
        "key_y": (int, "form"),
        "key_z": (str, "url")
    })
    def post(self):
        # logic code
        pass

在裝飾器中分別從不同的地方，form或者url中獲取相應的參數。如果取不到，則直接報錯，邏輯也不會進入到post函數中。

這是我基于flask這個框架自己總結出來的一套尚且還能看能用的參數解析方式，如果在每個函數中通過框架提供的get_argument來逐一獲取參數，則顯得太丑，而且每個接口所需要的數據是什么也不夠直觀。不過這種方式我自己還不是特別滿意，總感覺還是有點不太舒服，也說不清不舒服在哪里。那就干脆放棄它，使用別的方式吧。

后來我了解到了jsonschema這個東西，看了一下感覺與ProtocolBuffer很相似，只不過它是采用json的格式定義，正合我意(對于它我也有點吐槽，在數據庫層有提到)，每次數據進來就對數據和schema作一次validate操作，再進入業務邏輯層。

業務邏輯層的重構其實改動的代碼并不多，把一些同步的操作改成異步的操作。就拿如何重構某個接口來說吧，重構前的代碼可能是這樣的:

def function_before_refactor(some_params):
    result_1 = sync_call_1(some_params)
    result_2 = sync_call_2(some_params)
    # some other processes
    return result

使用gen.coroutine重構后:

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    # if you don"t want to refactor
    # just call it as it always be
    result_1 = sync_call_1(some_params)
    result_2 = yield async_call_2(some_params)
    # some other processes
    raise gen.Return(result)
    # python3及以上的版本不需要采用拋出異常的方式，直接return就可以了
    # return result

考慮到函數名根本不用改，重構的過程非常容易：

函數用gen.coroutine包裝成協程

已經重構成異步方式的函數調用時添加yield關鍵字即可

函數返回采用raise gen.Return(result)的方式（僅限于Python 2.7）

因為我目前采用的是python 2.7，所以在處理返回的時候要用拋出異常的方式，在這種方式下有一個點需要注意到，那就是與平常異常的處理的混用，不然會導致邏輯流執行混亂：

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    try:
        # some logic code
        pass
    except Exception as e:
        if isinstance(e, gen.Return):
            # return the value raised by logic
            raise gen.Return(e.value)
        # more exception process

數據庫采用的是mongodb，在flask框架中采用了mongoengine作為數據庫層的orm，對于這個python-mongodb的orm產品，我個人并不是很喜歡(可能是因為我習慣了mongoose的工作方式)，這里面嵌套json的定義居然不能體現在schema中，需要分開定義兩個schema，然后再作引入的操作。比如（代碼只是用作演示，與項目無關）：

class Comment(EmbeddedDocument):
    content = StringField()
    # more comment details

class Page(Document):
    comments = ListField(EmbeddedDocumentField(Comment))
    # more page details

而在mongoose中就直觀多了:

var PageSchema = new Schema({
    title       :   {type : String, required : true},
    time        :   {type : Date, default : Date.now(), required : true},
    comments    :   [{
        content   :   {type : String}
        // more comment details
    }]
    // more page details
});

扯遠了，在tornado的框架中，再使用mongoengine就不合適了，畢竟有著異步和同步的區別。那有什么比較好的python-mongodb的異步orm框架呢？搜了下，有一個叫做motorengine的東西，orm的使用方式和mongoengine基本一樣，但看它的star數實在不敢用呀。而且它處理異步的方式是使用回調，現在都是使用協程的年代了，想想還是算了吧。

最后找了個motor，感覺還不錯，它有對目前大部分主流協程框架的支持，操作mongodb的方式與直接使用pymongo的方式差不多（畢竟都是基于pymongo的封裝嘛），但是就是沒有orm的驗證層，那就自己再去另外搞一個簡化的orm層吧。(mongokit的orm方式看上去還不錯，但貌似對協程框架的支持一般)。這里暫時先懶惰一下，還是采用了jsonschema。每次保存前都validate一下對象是否符合schema的定義。如果沒有類mongoose的python-mongodb異步框架，有時間就自己寫一個吧~

這里順帶吐槽一下jsonschema，簡直太瑣碎了，一個很短的文檔結構定義，它會描述成好幾十行，我就不貼代碼了，有興趣的朋友可以戳這里http://jsonschema.net/玩玩。而且python中的jsonschema庫還不支持對于default關鍵字的操作，參見這個issue。

python中的測試框架有很多，只要選擇一個合適的能夠很方便與項目集成就好。我個人還是很喜歡unittest這個框架，小而精。我的這套測試方案也是基于unittest框架的。

# TestUserPostAccessComponents.py
class TestUserPostAccessComponents(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 定義在其它地方，具體細節就不展示了
        # 在setup中使用測試賬號獲取登陸態
        # 并把各種中間用得到的信息放在TestUserPostAccess類上
        setup(cls)

    @classmethod
    def tearDownClass(cls):
        pass

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_1_user_1_user_2_add_friend(self):
        pass

    def test_2_user_1_user_2_del_friend(self):
        pass

    def test_3_user_1_add_public_user_post(self):
        pass

    # more other components

最頂層的測試文件：

# run_test.py
# 各種import

def user_basic_post_access_test():
    tests = ["test_3_user_1_add_public_user_post",
             "test_5_user_2_as_a_stranger_can_access_public_user_post",
             "test_4_user_1_del_public_user_post",
             "test_6_user_1_add_private_user_post",
             "test_8_user_2_as_a_stranger_can_not_access_private_user_post",
             "test_9_user_1_self_can_access_private_user_post",
             "test_7_user_1_del_private_user_post"]
    return unittest.TestSuite(map(TestUserPostAccessComponents, tests))

def other_process_test():
    tests = [
        # compose a process by components by yourself
    ]
    return unittest.TestSuite(map(OtherTestCaseComponents, tests))

runner = unittest.TextTestRunner(verbosity=2)
runner.run(user_basic_post_access_test())
runner.run(other_process_test())

這套測試是基于 BDD (行為驅動)的測試方式，針對每一個邏輯模塊，定義一個components類，把所有子操作都定義成多帶帶的測試單元。這里面的測試單元可以是完全無序的，把邏輯有序化組織成測試用例的過程會在最外面通過TestSuit的方式組織起來。這里可能會有一些異議，因為有些人在使用這個測試類的時候是把它作為一個測試用例來組織的，當然這些都是不同的使用方式。

這套測試方案中的每個component都是api級別的測試，并不是函數級別的測試（集成測試與單元測試），每個TestSuit都是完整的一個業務流程。這樣的好處在于 測試和項目完全解耦。測試代碼不用關心項目的代碼是同步還是異步的。就算項目重構了，測試完全無感知，只要api沒變，就可以繼續工作。

當然以上都是理想的狀態，因為在剛開始寫這些測試的時候我還沒有總結到這些點，導致了一些耦合性的存在。比如說測試代碼中import了項目中的某個函數去獲取一些數據，用于檢查某個component的更新操作是否成功。在重構的過程中，該函數被重構成了協程。這樣一來，在測試代碼中就不能采用原來一樣的方式去調用了，也就是說測試代碼受到了框架同步與異步的影響，下一節我們就來談談同步與異步的測試，以及對于這種問題的解決方案。

在tornado中，也提供了一套測試的功能，具體在tornado.testing這個模塊，看它源碼其實可以發現它也是基于unittest的一層封裝。
我心里一直有一個問題：unittest的執行流程是同步的，既然這樣，它是怎么去測一個由gen.coroutine包裝的協程的呢，畢竟后者是異步的。
直到看了源碼，恍然大悟，原來是io_loop.run_sync這個函數的功勞，具體實現在gen_test這個裝飾器中，摘一部分源碼（對于tornado源碼不熟的同學可以先去看看tornado中的ioloop模塊的實現，看完會對這個部分有更深刻的理解）：

def gen_test(func=None, timeout=None):
    if timeout is None:
        timeout = get_async_test_timeout()

    def wrap(f):
        # Stack up several decorators to allow us to access the generator
        # object itself.  In the innermost wrapper, we capture the generator
        # and save it in an attribute of self.  Next, we run the wrapped
        # function through @gen.coroutine.  Finally, the coroutine is
        # wrapped again to make it synchronous with run_sync.
        #
        # This is a good case study arguing for either some sort of
        # extensibility in the gen decorators or cancellation support.
        @functools.wraps(f)
        def pre_coroutine(self, *args, **kwargs):
            result = f(self, *args, **kwargs)
            if isinstance(result, types.GeneratorType):
                self._test_generator = result
            else:
                self._test_generator = None
            return result

        coro = gen.coroutine(pre_coroutine)

        @functools.wraps(coro)
        def post_coroutine(self, *args, **kwargs):
            try:
                return self.io_loop.run_sync(
                    functools.partial(coro, self, *args, **kwargs),
                    timeout=timeout)
            except TimeoutError as e:
                # run_sync raises an error with an unhelpful traceback.
                # If we throw it back into the generator the stack trace
                # will be replaced by the point where the test is stopped.
                self._test_generator.throw(e)
                # In case the test contains an overly broad except clause,
                # we may get back here.  In this case re-raise the original
                # exception, which is better than nothing.
                raise
        return post_coroutine

    if func is not None:
        # Used like:
        #     @gen_test
        #     def f(self):
        #         pass
        return wrap(func)
    else:
        # Used like @gen_test(timeout=10)
        return wrap

在源碼中，先把某個測試單元封裝成一個協程，然后獲取當前線程的ioloop對象，把協程拋給他去執行，直到執行完畢。這樣就完美地實現了異步到同步的過渡，滿足unittest測試框架的同步需求。
在具體的使用中只需要繼承tornado提供的AsyncTestCase類就行了，注意這里不是unittest.TestCase。看了源碼也可以發現，前者就是繼承自后者的。

# This test uses coroutine style.
class MyTestCase(AsyncTestCase):
    @tornado.testing.gen_test
    def test_http_fetch(self):
        client = AsyncHTTPClient(self.io_loop)
        response = yield client.fetch("http://www.tornadoweb.org")
        # Test contents of response
        self.assertIn("FriendFeed", response.body)

回到上一節的問題，有了這種方式，就可以很容易地解決同步異步的問題了。如果測試用例中某一個函數已經被項目重構成了協程，只需要做以下三步：

把測試components的類改成繼承自AsyncTestCase

該測試單元使用gen_test裝飾（其它測試單元可以不用加，只需要改涉及到協程的測試單元就行）

調用協程的地方添加yield關鍵字

如果是api測試
測試中盡量不要調用任何項目中的代碼，它只專注于測試接口是否按照預期在工作，具體里面是怎么樣的不需要關心。這樣的話整套測試是完全獨立于項目而存在的，即使項目重構，也可以不用作任何修改，無縫對接。

如果是單元測試
參考上一節的方案。

重構是一個不斷優化和學習的過程，在這個過程中我踩了一些坑，也爬出了一些坑，希望可以把我的這些總結分享給大家。歡迎大家跟我交流。對于文中的一些方案，也歡迎大家拍磚，歡迎有更多的做法可以一起探討學習。另外，對于這個項目的重構，文章里面可能還少了一些更加直觀的性能測試，后面我會加上去，孝敬各位爺~