Django、Flask 和 Redis 教程：Python 框架之間的 Web 應用程序會話管理

已發表: 2022-03-11

Django 與 Flask：當 Django 是錯誤的選擇時

我喜歡並在我的許多個人和客戶項目中使用 Django，主要用於更經典的 Web 應用程序和那些涉及關係數據庫的應用程序。然而，Django 並不是靈丹妙藥。

按照設計，Django 與其 ORM、模板引擎系統和設置對象緊密耦合。另外，這不是一個新項目：它承載了很多包袱來保持向後兼容。

一些 Python 開發人員認為這是一個主要問題。他們說 Django 不夠靈活，盡可能避免使用它，而是使用像 Flask 這樣的 Python 微框架。

我不同意這種觀點。在適當的地點和時間使用 Django 非常棒，即使它不適合每個項目規範。正如口頭禪所說：“為工作使用正確的工具”。

（即使不是正確的地點和時間，有時使用 Django 編程也會有獨特的好處。）

在某些情況下，使用更輕量級的框架（如 Flask）確實會很好。通常，當您意識到這些微框架很容易被破解時，它們就會開始發光。

微框架的救援

在我的一些客戶項目中，我們討論過放棄 Django 並轉向微框架，通常是當客戶想做一些有趣的事情時（例如，在應用程序對像中嵌入 ZeroMQ）和項目使用 Django 似乎更難以實現目標。

更一般地說，我發現 Flask 可用於：

簡單的 REST API 後端
不需要數據庫訪問的應用程序
基於 NoSQL 的 Web 應用程序
具有非常特殊要求的 Web 應用程序，例如自定義 URL 配置

同時，我們的應用程序需要用戶註冊和 Django 多年前解決的其他常見任務。鑑於其重量輕，Flask 沒有配備相同的工具包。

問題出現了：Django 是一個孤注一擲的交易嗎？

問題出現了：Django 是一個孤注一擲的交易嗎？我們應該將它完全從項目中刪除，還是我們可以學習將它與其他微框架或傳統框架的靈活性結合起來？我們可以挑選我們想要使用的部分並避開其他部分嗎？

我們可以兩全其美嗎？我說是的，尤其是在會話管理方面。

（更不用說，有很多針對 Django 自由職業者的項目。）

現在是 Python 教程：共享 Django 會話

這篇文章的目標是將用戶身份驗證和註冊的任務委託給 Django，同時使用 Redis 與其他框架共享用戶會話。我可以想到一些這樣的場景會很有用：

您需要與 Django 應用程序分開開發 REST API，但希望共享會話數據。
您有一個特定組件可能需要稍後更換或出於某種原因向外擴展，並且仍需要會話數據。

在本教程中，我將使用 Redis 在兩個框架（在本例中為 Django 和 Flask）之間共享會話。在當前設置中，我將使用 SQLite 存儲用戶信息，但如果需要，您可以將後端綁定到 NoSQL 數據庫（或基於 SQL 的替代方案）。

了解會話

要在 Django 和 Flask 之間共享會話，我們需要了解一下 Django 是如何存儲其會話信息的。 Django 文檔非常好，但我將提供一些完整的背景知識。

會話管理品種

通常，您可以選擇通過以下兩種方式之一來管理 Python 應用程序的會話數據：

基於 Cookie 的會話：在這種情況下，會話數據不存儲在後端的數據存儲中。相反，它被序列化、簽名（使用 SECRET_KEY）並發送給客戶端。當客戶端發回該數據時，會檢查其完整性是否被篡改，並在服務器上再次對其進行反序列化。
基於存儲的會話：在這種情況下，會話數據本身不會發送到客戶端。相反，只發送一小部分（密鑰）來指示當前用戶的身份，存儲在會話存儲中。

在我們的示例中，我們對後一種情況更感興趣：我們希望將會話數據存儲在後端，然後在 Flask 中檢查。前者可以做同樣的事情，但正如 Django 文檔所述，第一種方法的安全性存在一些問題。

一般工作流程

會話處理和管理的一般工作流程將類似於此圖：

顯示使用 Redis 管理 Flask 和 Django 之間的用戶會話的圖表。

讓我們更詳細地了解會話共享：

當一個新的請求進來時，第一步是通過 Django 堆棧中註冊的中間件發送它。我們對SessionMiddleware類感興趣，正如您所料，它與會話管理和處理有關：
```
 class SessionMiddleware(object): def process_request(self, request): engine = import_module(settings.SESSION_ENGINE) session_key = request.COOKIES.get(settings.SESSION_COOKIE_NAME, None) request.session = engine.SessionStore(session_key)
```
在這個片段中，Django 獲取已註冊的SessionEngine （我們將很快了解），從request （默認為sessionid ）中提取SESSION_COOKIE_NAME並創建所選SessionEngine的新實例來處理會話存儲。

稍後（在處理用戶視圖之後，但仍在中間件堆棧中），會話引擎調用其保存方法來保存對數據存儲的任何更改。（在視圖處理期間，用戶可能在會話中更改了一些內容，例如，通過使用request.session向會話對象添加新值。）然後， SESSION_COOKIE_NAME被發送到客戶端。這是簡化版：

 def process_response(self, request, response): .... if response.status_code != 500: request.session.save() response.set_cookie(settings.SESSION_COOKIE_NAME, request.session.session_key, max_age=max_age, expires=expires, domain=settings.SESSION_COOKIE_DOMAIN, path=settings.SESSION_COOKIE_PATH, secure=settings.SESSION_COOKIE_SECURE or None, httponly=settings.SESSION_COOKIE_HTTPONLY or None) return response

我們對SessionEngine類特別感興趣，我們將用一些東西來代替它來存儲和加載數據到 Redis 後端。

幸運的是，有一些項目已經為我們處理了這個問題。這是redis_sessions_fork的一個例子。密切注意save和load方法，它們的編寫目的是（分別）將會話存儲到 Redis 和從 Redis 加載會話：

 class SessionStore(SessionBase): """ Redis session back-end for Django """ def __init__(self, session_key=None): super(SessionStore, self).__init__(session_key) def _get_or_create_session_key(self): if self._session_key is None: self._session_key = self._get_new_session_key() return self._session_key def load(self): session_data = backend.get(self.session_key) if not session_data is None: return self.decode(session_data) else: self.create() return {} def exists(self, session_key): return backend.exists(session_key) def create(self): while True: self._session_key = self._get_new_session_key() try: self.save(must_create=True) except CreateError: continue self.modified = True self._session_cache = {} return def save(self, must_create=False): session_key = self._get_or_create_session_key() expire_in = self.get_expiry_age() session_data = self.encode(self._get_session(no_load=must_create)) backend.save(session_key, expire_in, session_data, must_create) def delete(self, session_key=None): if session_key is None: if self.session_key is None: return session_key = self.session_key backend.delete(session_key)

了解這個類是如何運行的很重要，因為我們需要在 Flask 上實現類似的東西來加載會話數據。讓我們通過一個 REPL 示例仔細看看：

 >>> from django.conf import settings >>> from django.utils.importlib import import_module >>> engine = import_module(settings.SESSION_ENGINE) >>> engine.SessionStore() <redis_sessions_fork.session.SessionStore object at 0x3761cd0> >>> store["count"] = 1 >>> store.save() >>> store.load() {u'count': 1}

會話存儲的界面很容易理解，但背後有很多事情要做。我們應該再深入一點，以便我們可以在 Flask 上實現類似的東西。

注意：您可能會問，“為什麼不直接將 SessionEngine 複製到 Flask 中？” 說起來容易做起來難。 正如我們在開頭所討論的，Django 與其 Settings 對象緊密耦合，因此您不能只導入一些 Django 模塊並在沒有任何額外工作的情況下使用它。

Django 會話（反）序列化

正如我所說，Django 做了很多工作來掩蓋其會話存儲的複雜性。讓我們檢查存儲在上述代碼段中的 Redis 密鑰：

 >>> store.session_key u"ery3j462ezmmgebbpwjajlxjxmvt5adu"

現在，讓我們在 redis-cli 上查詢該鍵：

 redis 127.0.0.1:6379> get "django_sessions:ery3j462ezmmgebbpwjajlxjxmvt5adu" "ZmUxOTY0ZTFkMmNmODA2OWQ5ZjE4MjNhZmQxNDM0MDBiNmQzNzM2Zjp7ImNvdW50IjoxfQ=="

我們在這裡看到的是一個非常長的 Base64 編碼字符串。要了解它的用途，我們需要查看 Django 的SessionBase類來了解它是如何處理的：

 class SessionBase(object): """ Base class for all Session classes. """ def encode(self, session_dict): "Returns the given session dictionary serialized and encoded as a string." serialized = self.serializer().dumps(session_dict) hash = self._hash(serialized) return base64.b64encode(hash.encode() + b":" + serialized).decode('ascii') def decode(self, session_data): encoded_data = base64.b64decode(force_bytes(session_data)) try: hash, serialized = encoded_data.split(b':', 1) expected_hash = self._hash(serialized) if not constant_time_compare(hash.decode(), expected_hash): raise SuspiciousSession("Session data corrupted") else: return self.serializer().loads(serialized) except Exception as e: # ValueError, SuspiciousOperation, unpickling exceptions if isinstance(e, SuspiciousOperation): logger = logging.getLogger('django.security.%s' % e.__class__.__name__) logger.warning(force_text(e)) return {}

encode 方法首先使用當前註冊的序列化器對數據進行序列化。換句話說，它將會話轉換為字符串，然後可以將其轉換回會話（有關更多信息，請參閱 SESSION_SERIALIZER 文檔）。然後，它對序列化數據進行哈希處理，並在稍後使用此哈希作為簽名來檢查會話數據的完整性。最後，它將該數據對作為 Base64 編碼字符串返回給用戶。

順便說一句：在 1.6 版本之前，Django 默認使用 pickle 進行會話數據的序列化。 出於安全考慮，默認的序列化方法現在是django.contrib.sessions.serializers.JSONSerializer 。

編碼示例會話

讓我們看看實際的會話管理過程。在這裡，我們的會話字典將只是一個計數和一些整數，但您可以想像這將如何推廣到更複雜的用戶會話。

 >>> store.encode({'count': 1}) u'ZmUxOTY0ZTFkMmNmODA2OWQ5ZjE4MjNhZmQxNDM0MDBiNmQzNzM2Zjp7ImNvdW50IjoxfQ==' >>> base64.b64decode(encoded) 'fe1964e1d2cf8069d9f1823afd143400b6d3736f:{"count":1}'

store 方法的結果 (u'ZmUxOTY…==') 是一個包含序列化用戶會話及其哈希的編碼字符串。當我們解碼它時，我們確實得到了哈希（'fe1964e...'）和會話（ {"count":1} ）。

請注意，decode 方法會檢查以確保該會話的哈希是正確的，從而在我們在 Flask 中使用它時保證數據的完整性。在我們的例子中，我們不太擔心我們的會話在客戶端被篡改，因為：

我們沒有使用基於 cookie 的會話，也就是說，我們沒有將所有用戶數據發送到客戶端。
在 Flask 上，我們需要一個只讀的SessionStore ，它會告訴我們給定的密鑰是否存在並返回存儲的數據。

擴展到燒瓶

接下來，讓我們創建一個簡化版本的 Redis 會話引擎（數據庫）來使用 Flask。我們將使用相同的SessionStore （如上定義）作為基類，但我們需要刪除它的一些功能，例如檢查錯誤簽名或修改會話。我們對一個只讀的SessionStore更感興趣，它將加載從 Django 保存的會話數據。讓我們看看它是如何組合在一起的：

 class SessionStore(object): # The default serializer, for now def __init__(self, conn, session_key, secret, serializer=None): self._conn = conn self.session_key = session_key self._secret = secret self.serializer = serializer or JSONSerializer def load(self): session_data = self._conn.get(self.session_key) if not session_data is None: return self._decode(session_data) else: return {} def exists(self, session_key): return self._conn.exists(session_key) def _decode(self, session_data): """ Decodes the Django session :param session_data: :return: decoded data """ encoded_data = base64.b64decode(force_bytes(session_data)) try: # Could produce ValueError if there is no ':' hash, serialized = encoded_data.split(b':', 1) # In the Django version of that they check for corrupted data # I don't find it useful, so I'm removing it return self.serializer().loads(serialized) except Exception as e: # ValueError, SuspiciousOperation, unpickling exceptions. If any of # these happen, return an empty dictionary (ie, empty session). return {}

我們只需要load方法，因為它是存儲的只讀實現。這意味著您不能直接從 Flask 註銷；相反，您可能希望將此任務重定向到 Django。請記住，這裡的目標是管理這兩個 Python 框架之間的會話，從而為您提供更大的靈活性。

燒瓶會話

Flask 微框架支持基於 cookie 的會話，這意味著所有會話數據都被發送到客戶端，經過 Base64 編碼和加密簽名。但實際上，我們對 Flask 的會話支持並不是很感興趣。

我們需要的是獲取 Django 創建的會話 ID 並與 Redis 後端進行檢查，以便我們可以確定請求屬於預簽名用戶。總之，理想的過程是（這與上圖同步）：

我們從用戶的 cookie 中獲取 Django 會話 ID。
如果在 Redis 中找到會話 ID，我們將返回與該 ID 匹配的會話。
如果沒有，我們會將它們重定向到登錄頁面。

有一個裝飾器來檢查該信息並將當前user_id設置到 Flask 中的g變量中會很方便：

 from functools import wraps from flask import g, request, redirect, url_for def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): djsession_id = request.cookies.get("sessionid") if djsession_id is None: return redirect("/") key = get_session_prefixed(djsession_id) session_store = SessionStore(redis_conn, key) auth = session_store.load() if not auth: return redirect("/") g.user_id = str(auth.get("_auth_user_id")) return f(*args, **kwargs) return decorated_function

在上面的示例中，我們仍然使用我們之前定義的SessionStore從 Redis 獲取 Django 數據。如果會話有_auth_user_id ，我們從視圖函數返回內容；否則，用戶將被重定向到登錄頁面，就像我們想要的那樣。

把東西粘在一起

為了共享 cookie，我發現通過 WSGI 服務器啟動 Django 和 Flask 並將它們粘合在一起很方便。在這個例子中，我使用了 CherryPy：

 from app import app from django.core.wsgi import get_wsgi_application application = get_wsgi_application() d = wsgiserver.WSGIPathInfoDispatcher({ "/":application, "/backend":app }) server = wsgiserver.CherryPyWSGIServer(("127.0.0.1", 8080), d)

這樣，Django 將在“/”上提供服務，而 Flask 將在“/backend”端點上提供服務。

綜上所述

我沒有研究 Django 與 Flask 的對比，也沒有鼓勵你只學習 Flask 微框架，而是將 Django 和 Flask 焊接在一起，通過將任務委託給 Django 來讓它們共享相同的會話數據以進行身份驗證。由於 Django 附帶了大量模塊來解決用戶註冊、登錄和註銷（僅舉幾例），將這兩個框架結合起來將為您節省寶貴的時間，同時為您提供破解像 Flask 這樣可管理的微框架的機會。