starsessions 2.1.0

明星会议

Starlette 和 FastAPI 框架的高级课程

安装

starsessions使用 PIP 或诗歌安装：

pip install starsessions
# or
poetry add starsessions

使用redis额外的Redis 支持。

快速开始

examples/请参阅此存储库目录中的示例应用程序。

用法

1. 添加starsessions.SessionMiddleware到您的应用程序以启用会话支持，
2. 配置会话存储并将其传递给中间件，
3. load_session(connection)通过调用实用程序在您的视图/中间件中加载会话。

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.responses import JSONResponse
from starlette.routing import Route

from starsessions import CookieStore, load_session, SessionMiddleware


async def index_view(request):
    await load_session(request)

    session_data = request.session
    return JSONResponse(session_data)


session_store = CookieStore(secret_key='TOP SECRET')

app = Starlette(
    middleware=[
        Middleware(SessionMiddleware, store=session_store, lifetime=3600 * 24 * 14),
    ],
    routes=[
        Route('/', index_view),
    ]
)

Cookie 安全性

默认情况下，中间件使用严格的默认值。cookie 生命周期仅限于浏览器会话，并且仅通过 HTTPS 协议发送。您可以通过更改cookie_https_only和lifetime参数来更改这些默认值：

from starlette.middleware import Middleware

from starsessions import CookieStore, SessionMiddleware

session_store = CookieStore(secret_key='TOP SECRET')

middleware = [
    Middleware(SessionMiddleware, store=session_store, cookie_https_only=False, lifetime=3600 * 24 * 14),
]

上面的示例将允许会话使用不安全的 HTTP 传输，并且会话生命周期将设置为 14 天。

加载会话

默认情况下不加载会话数据。调用load_session以从存储中加载数据。

async def index_view(request):
    await load_session(request)
    request.session['key'] = 'value'

但是，如果您尝试访问未初始化的会话，SessionNotLoaded则会引发异常。

async def index_view(request):
    request.session['key'] = 'value'  # raises SessionNotLoaded

您可以使用SessionAutoloadMiddleware中间件自动加载会话。

会话自动加载

出于性能原因，默认情况下不会自动加载会话。load_session有时打电话太频繁很烦人。我们SessionAutoloadMiddleware通过为您自动加载会话来减少样板代码的数量。

有两个选项：始终自动加载或仅针对特定路径自动加载。以下是示例：

from starlette.middleware import Middleware

from starsessions import CookieStore, SessionAutoloadMiddleware, SessionMiddleware

session_store = CookieStore(secret_key='TOP SECRET')

# Always autoload

middleware = [
    Middleware(SessionMiddleware, store=session_store),
    Middleware(SessionAutoloadMiddleware),
]

# Autoload session for selected paths

middleware = [
    Middleware(SessionMiddleware, store=session_store),
    Middleware(SessionAutoloadMiddleware, paths=['/admin', '/app']),
]

# regex patterns also supported
import re

admin_rx = re.compile('/admin*')

middleware = [
    Middleware(SessionMiddleware, store=session_store),
    Middleware(SessionAutoloadMiddleware, paths=[admin_rx]),
]

滚动会议

的默认行为是在设置后几秒钟后SessionMiddleware使 cookie 过期。lifetime例如，如果您创建一个会话，lifetime=3600则该会话将在 3600 秒内准确终止。有时这可能不是您所需要的，因此我们提供备用到期策略 - 滚动会话。

在使用滚动会话时，cookie 过期时间将lifetime在每个响应上按值延长。让我们看看它是如何在示例中工作的。首先，在第一个响应中，您创建一个新会话lifetime=3600，然后用户执行另一个请求，会话再延长 3600 秒，依此类推。当您想要进行短时会话但不希望它们在用户操作过程中中断时，这种方法很有用。使用滚动策略，会话 cookie 只有在用户一段时间不活动后才会过期。

启用滚动策略集rolling=True。

from starlette.middleware import Middleware
from starsessions import SessionMiddleware

middleware = [
    Middleware(SessionMiddleware, lifetime=300, rolling=True),
]

上面的代码片段演示了一个示例设置，其中会话将在 300 秒（5 分钟）不活动后被丢弃，但在用户在线时会自动延长 5 分钟。

饼干路径

您可以传递cookie_path参数以将会话 cookie 绑定到特定 URL。例如，要仅为管理区域激活会话 cookie，请使用cookie_path="/admin"中间件参数。

from starlette.middleware import Middleware
from starsessions import SessionMiddleware

middleware = [
    Middleware(SessionMiddleware, cookie_path='/admin'),
]

所有其他不匹配值的 URLcookie_path将不会收到 cookie，因此会话将不可用。

Cookie 域

您还可以通过将cookie_domain参数传递给中间件来指定哪些主机可以接收 cookie。

from starlette.middleware import Middleware
from starsessions import SessionMiddleware

middleware = [
    Middleware(SessionMiddleware, cookie_domain='example.com'),
]

请注意，这使得会话 cookie 也可用于子域。例如，当您设置时，cookie_domain=example.com会话 cookie 将在子域（如app.example.com.

仅会话 cookie

如果您希望会话 cookie 在选项卡关闭时自动从浏览器中删除，则设置lifetime为0.

注意，这取决于浏览器的实现！

from starlette.middleware import Middleware
from starsessions import SessionMiddleware

middleware = [
    Middleware(SessionMiddleware, lifetime=0),
]

内置商店

记忆

班级：starsessions.InMemoryStore

只需将数据存储在内存中。服务器重启后数据被清除。主要用于单元测试。

饼干店

班级：starsessions.CookieStore

将会话数据存储在客户端的签名 cookie 中。

雷迪斯

班级：starsessions.stores.redis.RedisStore

将会话数据存储在 Redis 服务器中。商店接受连接 URL 或Redis.

需要redis-py，使用pip install starsessions[redis]或poetry add starsessions[redis]

from redis.asyncio.utils import from_url

from starsessions.stores.redis import RedisStore

store = RedisStore('redis://localhost')
# or
redis = from_url('redis://localhost')

store = RedisStore(connection=redis)

Redis 键前缀

默认情况下，Redis 中的所有键都以starsessions.. 如果要更改此使用prefix参数。

from starsessions.stores.redis import RedisStore

store = RedisStore(url='redis://localhost', prefix='my_sessions')

前缀可以是可调用的：

from starsessions.stores.redis import RedisStore


def make_prefix(key: str) -> str:
    return 'my_sessions_' + key


store = RedisStore(url='redis://localhost', prefix=make_prefix)

密钥过期

该库自动管理密钥过期，通常您与它无关。但是对于lifetime=0我们不知道会话何时结束的情况，我们必须启发式地计算 TTL，否则数据将永远保留在 Redis 中。此时，我们只是设置了 30 天的 TTL。您可以通过gc_ttl在商店中设置值来更改它。

from starsessions.stores.redis import RedisStore

store = RedisStore(url='redis://localhost', gc_ttl=3600)  # max 1 hour

定制商店

创建新商店非常简单。您所需要的只是扩展starsessions.SessionStore 类并实现抽象方法。

这是我们如何创建基于内存的会话存储的示例。请注意，该write方法将会话 ID 作为字符串值返回很重要。

from typing import Dict

from starsessions import SessionStore


# instance of class which manages session persistence

class InMemoryStore(SessionStore):
    def __init__(self):
        self._storage = {}

    async def read(self, session_id: str, lifetime: int) -> Dict:
        """ Read session data from a data source using session_id. """
        return self._storage.get(session_id, {})

    async def write(self, session_id: str, data: Dict, lifetime: int, ttl: int) -> str:
        """ Write session data into data source and return session id. """
        self._storage[session_id] = data
        return session_id

    async def remove(self, session_id: str):
        """ Remove session data. """
        del self._storage[session_id]

    async def exists(self, session_id: str) -> bool:
        return session_id in self._storage

生命周期和 ttl

write接受两个特殊参数：lifetime和ttl。区别在于lifetime总会话持续时间（由中间件设置）和ttl剩余会话时间。几秒钟后ttl，数据可以安全地从存储中删除。

您的自定义后端必须在lifetime = 0. 在这种情况下，您没有确切的到期值，并且您必须找到一种方法如何在存储端扩展会话 TTL（如果有）。

序列化器

该库使用 JSON 自动将会话数据序列化为字符串。默认情况下，我们使用，但您可以通过扩展 类starsessions.JsonSerializer来实现自己的。starsessions.Serializer

import json
import typing

from starlette.middleware import Middleware

from starsessions import Serializer, SessionMiddleware


class MySerializer(Serializer):
    def serialize(self, data: typing.Any) -> bytes:
        return json.dumps(data).encode('utf-8')

    def deserialize(self, data: bytes) -> typing.Dict[str, typing.Any]:
        return json.loads(data)


middleware = [
    Middleware(SessionMiddleware, serializer=MySerializer()),
]

会话终止

如果会话没有数据，中间件将删除会话数据和 cookie。用于request.session.clear清空数据。

重新生成会话 ID

有时您需要一个新的会话 ID 来避免会话固定攻击（例如，在成功登录后）。为此，请使用starsessions.session.regenerate_session_id(connection)实用程序。

from starsessions.session import regenerate_session_id
from starlette.responses import Response


def login(request):
    regenerate_session_id(request)
    return Response('successfully signed in')