Python-FastAPI 異步框架開發博客系統（三）--異步特性篇

項目地址

博客地址

異步篇最接近Frodo的初衷了。通訊與數據的內容使用傳統框架的思路是相同的。而異步思路只改變了若干場景的實現方法。

異步編程不是新鮮概念，但他並無指定很明確的技術特色和路線。相關概念也不是很清晰，不多有文章能細緻地說明白 阻塞/非阻塞、異步/同步、並行/併發、分佈式、IO多路複用、協程 這些概念的區別與聯繫。這些概念在CS專業的OS、分佈式系統課程中可能有設計，但具體實現層面可能鮮有涉及。具體到Python這門語言，我閱讀了不少工業界、python屆的工做者（或者稱爲pythonista們）寫的文章，下面兩篇是最值得閱讀的：

小白的 asyncio ：原理、源碼 到實現（1） - 閒談後的文章 - 知乎； 固然標題是做者在自謙。該文做者結合CPython中asyncio標準源碼、函數棧幀的源碼和python函數上下文源碼實現講述了python異步的設計原理，並手寫了一個簡易版的事件循環和asyncio-future對象。

深刻理解 Python 異步編程（上）；這篇文章寫於2017年，當時asyncio還沒成爲標準庫。這篇文章大篇幅使用python和linux的epoll接口一步步實現了單線程異步IO，最後引出了asyncio的事件循環，證明了其便捷性。做者規劃還有中下篇講述asyncio的原理，但是目前還沒等到下文。做者安放文章代碼的倉庫已經累計了數十條催更的issue。

基本問題

還記得咱們再「通訊篇」繪製的時序圖嗎？用它表示一次用戶執行的邏輯是沒問題的，但實際實現中，咱們真的能這樣寫代碼嗎？這裏有兩個基本問題：

• 併發訪問問題，如何實現多人同時訪問你的博客web進程？
• 如何避免io阻塞，從而充分利用cpu的時間片？

第一個問題作過web開發的都很熟悉了，他的解決方案不少，由於這是軟件發展中必須面對的問題：

• os級別，io多路複用機制，成熟的爲linux的epoll機制，nginx即是基於此實現訪問併發。
• 編程語言使用多線程解決，以Flask爲例，使用本地線程解決線程安全問題。
• 編程語言使用異步編程解決，以nodejs爲例，promise+回調的方式。python就是以asyncio爲表明的異步生態圈。

第二個問題其實跟第一個問題是一個意思，把對象換成cpu便可。Frodo解決第一個問題使用的是相似asyncio事件循環的uvloop循環，他包裝成了一個機遇ASGI協議的web服務器uvicorn,他能夠啓動多個ASGI標準寫的app，內置一套事件循環實現併發訪問。

uvicorn main:app --reload --host 0.0.0.0 --port 8001

重點是Frodo對於第二個問題的解決，這些都是在程序細節中體現出的。

問題分析：哪裏存在IO阻塞

咱們拿「通訊篇」中CRUD的通訊邏輯舉例，咱們先標註出IO阻塞的地方, 而後對應到程序設計中的環節，再來思考在實現中怎麼解決。

圖中標註出了三類io場景，並有的是串行的需求，有的是併發（能夠併發）的需求。我來分別解釋下：

• 第一類： 網絡的鏈接和斷開，http是基於tcp的可靠傳輸協議，創建鏈接的過程也是耗時的io操做。數據庫的鏈接是網絡鏈接或套接字文件讀寫類的連接，也是io耗時的。這些代碼主要在web中的checkpoin函數，在Frodo的views目錄下。
• 第二類： 通訊異步是指客戶端發送請求，等待數據準備好到返回的過程，這部分等到的時間實際上是後端的數據io操做，cpu不該被這段時間佔用。這部分代碼在Frodo的mdoels下。
• 第三類： 數據異步是指跟數據庫操做等待數據返回所需的時間消耗。這部分時間也應該還給cpu。

上述的不少場景必須是串行完成的，好比創建數據庫鏈接-->數據操做-->斷開鏈接。也有一些場景（主要是不涉及數據一致性的場景）能夠是並行的，如緩存的更新與刪除，由於KV數據庫不涉及關係的聯立，能夠並行地刪除。

解決方案

第一類：鏈接耗時

數據庫的鏈接與退出同步中都會想到使用帶with關鍵字的鏈接池，異步爲了這一鏈接過程能夠「被等待」或者說交出執行權給主程序，須要使用async關鍵字包裝一下，並實現異步上下文的方法__aenter__, __aexit__.

import databases

class AioDataBase():
    async def __aenter__(self):
        db = databases.Database(DB_URL.replace('+pymysql', ''))
        await db.connect()
        self.db = db
        return db

    async def __aexit__(self, exc_type, exc, tb):
        if exc:
            traceback.print_exc()
        await self.db.disconnect()

事實上，aiomysql已經幫助咱們實現了相似的功能，但很遺憾aiomysql不能和sqlalchemy配套使用，database是一個簡單的異步的數據庫驅動引擎，能執行sqlalchemy生成的sql。

第二類：通訊耗時

這點可否異步直覺決定了web應用的響應速度，異步下的checkpoint函數自己爲async def 關鍵字的協程，再由uvloop調度。對於此類函數的要求是對於阻塞操做一概使用await等待，看個例子：

@app.post('/auth')
async def login(req: Request, username: str=Form(...), password: str=Form(...)):
    user_auth: schemas.User = \
            ## 涉及到IO的函數須要等待
            await user.authenticate_user(username, password)
    if not user_auth:
        raise HTTPException(status_code=400, 
                            detail='Incorrect User Auth.')
    access_token_expires = timedelta(
        minutes=int(config.ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    access_token = await user.create_access_token(
                        data={'sub': user_auth.name},
                        expires_delta=access_token_expires)
    return { ... }

async def authenticate_user(
        username: str, password: str) -> schemas.User:
    user = await User.async_first(name=username)
    user = schemas.UserAuth(**user)
    if not user: return False
    if not verify_password(password, user.password): return False
    return user

你可能注意到了有些函數如verify_password並無等待他，由於他是計算任務，不可被等待。咱們只需按照邏輯把io耗時操做等待便可。

第三類：數據操做耗時

這體如今異步ORM方法的設計上，database + sqlalchemy的實現範例以下：

@classmethod
async def asave(cls, *args, **kwargs):
    '''  update  '''
    table = cls.__table__
    id = kwargs.pop('id')
    async with AioDataBase() as db:
        query = table.update().\
                        where(table.c.id==id).\
                        values(**kwargs)
        ## 等待1： 執行sql語句
        rv = await db.execute(query=query)
    ## 等待2： 拿取數據構造對象
    obj = cls(**(await cls.async_first(id=id)))
    ## 等待3： 清除對象涉及的緩存
    await cls.__flush__(obj)
    return rv

以更新數據數據爲例，涉及到的等待。同步的ORM框架像pymysql在db.execute(...)這類方法上式不能夠被等待的，直接是阻塞的，異步的寫法裏要等待他的結果，帶來的好處即是等待的時間執行權歸還主程序，使其能夠處理其餘事務。

並行的實現

異步下的並行是指不少io操做並不涉及數據一致性，能夠並行處理，好比刪除沒有關係的數據，查詢若干數據，更新沒有關係的數據等，這些均可以並行。異步中也容許這些並行，藉助asycio.gather(*coros)方法實現，這個方法將傳遞進去的協程都放入事件循環隊列，逐個執行相似coro.send(None)的操做，由於協程立馬退出，因此全部協程能夠立馬「同時」被喚醒等待，達到並行的效果。

類設計中使用的tricks

本節的內容是在使用python異步中的一些小技巧，能夠幫助咱們實現更好的設計。

將類的@property屬性序列化

序列化對象很常見，尤爲是想在緩存中存儲對象時須要序列化。對象的有些屬性是用異步@property完成的，跟其餘屬性不一樣，他們須要特殊的調用：

class Post(BaseModel):
    ...
    @property
    async def html_content(self):
        content = await self.content
        if not content:
            return ''
        return markdown(content)

這個property有些是異步的，每次使用此屬性時都須要content = await post.html_content, 而不帶async和await的屬性能夠直接訪問content = post.html_content。

這就給咱們的序列化方法帶來了麻煩。 咱們想讓類擁有一個知道本身有哪些異步property的功能，從而能在BaseModel中實現統一的序列化方法（在子類分別實現序列化方法是不現實的）。

讓類附加一個partials的屬性，存儲須要等待的property， 對於python，控制類的行爲（注意是類的建立行爲，不是實例的建立行爲）須要改變其元類，咱們設計一個叫PropertyHolder的元類，讓他的行爲控制全部數據類的生成：

class PropertyHolder(type):
    """
    We want to make our class with som useful properties 
    and filter the private properties.
    """
    def __new__(cls, name, bases, attrs):
        new_cls = type.__new__(cls, name, bases, attrs)
        new_cls.property_fields = []

        for attr in list(attrs) + sum([list(vars(base))
                                       for base in bases], []):
            if attr.startswith('_') or attr in IGNORE_ATTRS:
                continue
            if isinstance(getattr(new_cls, attr), property):
                new_cls.property_fields.append(attr)
        return new_cls

他的功能是過濾出咱們所須要的@property, 直接付給類的properties屬性。

接下來就是改變BaseModel的生成元類：

@as_declarative()
class Base():
    __name__: str
    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

    @property
    def url(self):
        return f'/{self.__class__.__name__.lower()}/{self.id}/'

    @property
    def canonical_url(self):
        pass

class ModelMeta(Base.__class__, PropertyHolder):
    ...


class BaseModel(Base, metaclass=ModelMeta):
    ...

Base是ORM的基類，他自己的元類也被改變（意味着不是type）,若是直接改變它則會讓咱們的數據類型喪失ORM的功能，一箭雙鵰的辦法是建立一個新的類同時繼承Base和PropertyHolder, 使這個類成爲新的混合元類。（_好繞啊，這裏的套娃現象我也不想的，我會慢慢找到更好的方案的..._）。

tricks: 類的元類如何拿到？ 調用 cls.__class__ 獲取他基於的元類。記住，python中類自己也是對象。他的建立也是受控制的。

關於fastapi

好了，Frodo第一個版本的核心設計思路已經介紹完了，前面的敘述中，我不多提fastapi，由於異步web自己和框架是不要緊的，這套內容換成sanic,aiohttp,tornado甚至是Django都是同樣的，只是具體的實現手段不一樣，好比Django的異步是基於他本身設計的channel實現的。

但fastapi也有他的特別之處，設計思想兼容幷蓄，也思考了不少，在開發中我強烈推薦使用的幾個地方：

• 數據模式schema的設計，配套pydantic的類型檢查，讓python這門動態語言變得更加可讀、調試更加容易、語法更加規範，我相信這是將來的趨勢。
• Depends的設計，咱們曾想過把複用的邏輯封裝成類、函數、裝飾器，但fastapi直接在參數上作文章，令我驚訝，他在參數上就代替了上下文、多參數、表單參數、認證參數等。
• 兼容同步寫法，包含WSGI，使用同步的技術庫搭配fastapi徹底沒問題，他容許同步函數的存在，緣由即是他基於的ASGI認爲本身是WSGI的超集，應當兼容兩種寫法。
• 配套swagger-doc, 後端福利，使得你不須要花費時間學習OpenAPI 語法即可順利作出先後端人員都能用、都能理解的調試平臺和文檔，省時省力。

Frodo的三篇介紹到此就完結了，靠課餘、科研時間以外的空隙完成的項目不免漏洞百出。但一個月的戰線後總算是完成了第一個版本。將來的目標是星辰大海，新語言的加入、多服務的拆分、虛擬化部署都須要時間的檢驗，努力吧~！