【FastAPI 學習 十】使用Redis

在FastAPI中 使用Redis

FastAPI官網關於異步的解釋描述 https://fastapi.tiangolo.com/async/
建議要使用FastAPI的人，都看看作者關於異步的描述

思路

把redis_cli對象掛載到FastAPI app 對象上面，然后在視圖函數中使用默認的回調參數request對象獲取

2021年1月30號更新 直接使用全局redis客戶端對象，實現方式在最下面。

參考鏈接
https://github.com/tiangolo/fastapi/issues/1694
https://github.com/tiangolo/fastapi/issues/1742
https://github.com/leonh/redis-streams-fastapi-chat/blob/master/chat.py

測試代碼

安裝aioredis

pip intsall aioredis

完整代碼，新建一個test_aioredis.py文件

from aioredis import create_redis_pool, Redis

from fastapi import FastAPI, Request, Query

app = FastAPI()


async def get_redis_pool() -> Redis:
    redis = await create_redis_pool(f"redis://:root12345@172.16.137.129:6379/0?encoding=utf-8")
    return redis


@app.on_event('startup')
async def startup_event():
    """
    獲取鏈接
    :return:
    """
    app.state.redis = await get_redis_pool()


@app.on_event('shutdown')
async def shutdown_event():
    """
    關閉
    :return:
    """
    app.state.redis.close()
    await app.state.redis.wait_closed()


@app.get("/test", summary="測試redis")
async def test_redis(request: Request, num: int=Query(123, title="參數num")):
    # 等待redis寫入  await異步變同步 如果不關心結果可以不用await，但是這里下一步要取值，必須得先等存完值 后再取值
    await request.app.state.redis.set("aa", num)
    # 等待 redis讀取
    v = await request.app.state.redis.get("aa")
    print(v, type(v))
    return {"msg": v}


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='test_aioredis:app', host="127.0.0.1", port=8080, reload=True, debug=True)

上面只是一個文件， 如何在項目中組織了？ 很多博客，都是單文件演示，不夠友好。

FastAPI項目中組織

仿照 flask 注冊掛載redis

def create_app():
    """
    生成FatAPI對象
    :return:
    """
    app = FastAPI()

    # 其余的一些全局配置可以寫在這里 多了可以考慮拆分到其他文件夾
    # 跨域設置
    # 注冊路由
    # 注冊捕獲全局異常
    # 請求攔截

    # 掛載redis
    register_redis(app)
    return app

def register_redis(app: FastAPI) -> None:
    """
    把redis掛載到app對象上面
    :param app:
    :return:
    """

    @app.on_event('startup')
    async def startup_event():
        """
        獲取鏈接
        :return:
        """
        app.state.redis = await create_redis_pool(settings.REDIS_URL)

    @app.on_event('shutdown')
    async def shutdown_event():
        """
        關閉
        :return:
        """
        app.state.redis.close()
        await app.state.redis.wait_closed()

使用

這個就和上面例子一樣，直接使用。

@app.get("/test", summary="測試redis")
async def test_redis(request: Request, num: int=Query(123, title="參數num")):
    # 等待redis寫入  await異步變同步 如果不關心結果可以不用await，但是這里下一步要取值，必須得先等存完值 后再取值
    await request.app.state.redis.set("aa", num)
    # 等待 redis讀取
    v = await request.app.state.redis.get("aa")
    print(v, type(v))
    return {"msg": v}

更改初始化位置

2021年1月30號更新

之前是吧 redis 對象放在了 請求對象上面，有時候就感覺特別不合理，於是就又找了個方式 直接全局引用。

Python不像Go可以通過指針直接修改原來的對象，但是可以通過class 實例化對象可以直接修改內部屬性的特性
再通過魔法方法，賦予實例化對象 具有內部屬性_redis_client的方法和屬性，就達到了上述的目的了。

具體代碼如下:

import sys
import redis
from common.logger import logger
from core.config import settings


class RedisCli(object):

    def __init__(self, *, host: str, port: int, password: str, db: int, socket_timeout: int = 5):
        # redis對象 在 @app.on_event("startup") 中連接創建
        self._redis_client = None
        self.host = host
        self.port = port
        self.password = password
        self.db = db
        self.socket_timeout = socket_timeout

    def init_redis_connect(self) -> None:
        """
        初始化連接
        :return:
        """
        try:
            self._redis_client = redis.Redis(
                host=self.host,
                port=self.port,
                password=self.password,
                db=self.db,
                socket_timeout=5,
                decode_responses=True  # 解碼
            )
            if not self._redis_client.ping():
                logger.info("連接redis超時")
                sys.exit()
        except (redis.AuthenticationError, Exception) as e:
            logger.info(f"連接redis異常 {e}")
            sys.exit()

    # 使實例化后的對象 賦予redis對象的的方法和屬性
    def __getattr__(self, name):
        return getattr(self._redis_client, name)

    def __getitem__(self, name):
        return self._redis_client[name]

    def __setitem__(self, name, value):
        self._redis_client[name] = value

    def __delitem__(self, name):
        del self._redis_client[name]


# 創建redis連接對象 但是這種方式使用方法時沒有提示
redis_client = RedisCli(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    password=settings.REDIS_PASSWORD,
    db=settings.REDIS_DB
)

# 只允許導出 redis_client 實例化對象
__all__ = ["redis_client"]

然后在 redis對象 在 @app.on_event("startup") 中使用init_redis_connect方法創建連接即可，然后其他地方可直接引用這個redis對象操作，但是不方便的由於Python沒有強制指定類型這一點，所以使用時沒有方法提示。

總結

最后推薦每個使用FastAPI的開發者都看下這個
FastAPI官網關於異步的解釋描述 https://fastapi.tiangolo.com/async/

話說這個async await語法糖和js ES6特別像

具體完整GitHub代碼

見個人網站https://www.charmcode.cn/article/2020-07-29_fastapi_redis