Litestar Stores 存储指南内置键值存储、命名空间与 StoreRegistry 统一管理【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar导读Litestar 内置了一套低层级的异步键值存储key/value store体系专门用于解决应用中简单存储的需求例如 响应缓存、服务端会话Server-side sessions等场景——在这些地方通常不需要引入完整的数据库一个线程安全、进程安全的键值存储就足够了。本文以 docs/usage/stores.rst 为主体结合 litestar/stores 源码与 docs/examples/stores 中的全部示例系统讲解四种内置存储的选型、get/set/delete基础操作、过期与续期机制、命名空间隔离以及通过StoreRegistry实现全应用统一存取和与中间件、响应缓存等集成组件的深度配置。一、为什么需要 Store轻量存储的适用场景在开发 Web 应用时很多数据并不需要落进关系型数据库。典型场景包括缓存响应数据把计算开销较大的响应结果临时存起来参见 docs/usage/caching.rst服务端会话把会话数据保存在服务端而不是 Cookie 中参见ServerSideSessionConfig相关用法限流计数RateLimitMiddleware内部也通过 store 保存请求计数。在这些场景中一个简单的键值存储就完全够用。Litestar 因此提供了多个低层级的键值存储实现它们都通过异步接口工作并面向线程安全与进程安全设计。这些 store 由一个集中的 StoreRegistry 统一管理从而让整个应用以及第三方插件都能方便地存取同一个存储实例。从源码看所有 store 都继承自 Store 抽象基类它定义了统一的异步接口set、get、delete、delete_all、exists、expires_in并支持async with上下文管理协议__aenter__/__aexit__。二、四种内置 Store 的选型对比MemoryStore默认的内存存储MemoryStore 是最简单的实现内部用一个字典dictionary持有数据无持久化进程重启后数据即丢失非线程/多进程安全不适合多 worker 进程间共享数据开销最低非常适合基本应用场景例如缓存是 Litestar 内部的默认存储注册表默认工厂返回的正是MemoryStore实例见 registry.py 的 default_default_factory。注意如果你计划启用多个 Web worker 且需要跨进程通信应改用下面几种非内存存储之一否则不同进程各自持有独立的内存字典数据无法互通。FileStore基于文件系统的持久化存储FileStore 把数据以文件形式保存到磁盘内置持久化数据易于提取和备份速度低于内存方案适合数据量大、生命周期长或持久性要求高的场景支持namespacing命名空间。RedisStore基于 Redis 的存储RedisStore 以 Redis 为后端继承了 Redis 的原子性、过期机制、SCAN等全部能力几乎适用于所有应用场景支持命名空间。使用示例见 docs/examples/stores/namespacing.py。ValkeyStoreRedis 的替代后端ValkeyStore 以 Valkey 为后端——Valkey 是 Redis 因许可证变更而产生的分支。它与RedisStore能力对等同样适用于几乎所有应用并支持命名空间。在写作本文时valkey.asyncio.Valkey与redis.asyncio.Redis等价因此文档中关于 Redis 的所有说明同样适用于 Valkey。为什么不支持 Memcached文档明确说明Memcached 不是受支持的后端未来也不太可能加入。原因很直接Memcached 缺少一些基础能力——例如无法检查一个 key 的剩余过期时间也没有类似 RedisSCAN这样的命令来实现基于模式的键批量删除导致难以在它之上正确实现 store 所需的完整语义。三、Store 的基础操作get / set / delete一个 store 最基本的操作有三个方法作用get读取已存储的值set向 store 写入一个值delete删除一个已存储的值完整示例如 docs/examples/stores/get_set.pyfrom litestar.stores.memory import MemoryStore store MemoryStore() async def main() - None: value await store.get(key) print(value) # 输出 None因为该 key 尚未写入 await store.set(key, bvalue) value await store.get(key) print(value)几点实现细节依据 base.py 的抽象方法签名set(key, value, expires_inNone)value接受str | bytes字符串会先被 UTF-8 编码再存储expires_in接受秒数int或timedelta。get(key, renew_forNone)返回bytes | None。renew_for参数用于在读取时续期见下文续期小节。delete(key)删除不存在的 key 时是空操作no-op不会抛错。四、过期时间expires_in 与续期机制4.1 设置过期时间set方法带有一个可选参数expires_in用来指定值在多长时间后过期。示例见 docs/examples/stores/expiry.pyfrom asyncio import sleep from litestar.stores.memory import MemoryStore store MemoryStore() async def main() - None: await store.set(foo, bbar, expires_in1) value await store.get(foo) print(value) await sleep(1) value await store.get(foo) # 1 秒后已过期这里输出 None print(value)4.2 过期值的处理方式因 store 而异具体如何处理过期值由各 store 自行决定实现可能不同RedisStore使用 Redis 原生的过期机制TTL自动清理FileStore只在访问该值、或显式调用delete_expired方法时才删除过期值。4.3 每次访问续期适合会话与 LRU 缓存get方法的renew_for参数可以在每次读取时延长值的存活时间这对服务端会话、LRU最近最少使用缓存这类活跃即保活的场景非常有用。示例见 docs/examples/stores/expiry_renew_on_get.pyfrom asyncio import sleep from litestar.stores.memory import MemoryStore store MemoryStore() async def main() - None: await store.set(foo, bbar, expires_in1) await sleep(0.5) await store.get(foo, renew_for1) # 把剩余存活时间重置为 1 秒 await sleep(1) # 距写入已过去 1.5 秒原本已超过 1 秒的生命周期 # 但由于读取时续期了 1 秒值仍然可用 value await store.get(foo) print(value)根据 base.py 的文档说明只有当值最初设置了过期时间时renew_for才会真正续期若值未设置过期时间该参数是空操作no-op。4.4 手动清理过期值使用MemoryStore或FileStore时过期数据不会自动删除只会在以下两种时机被清理数据被访问时显式调用MemoryStore.delete_expired或FileStore.delete_expired时。因此文档建议定期调用delete_expired避免存储的数据无限增长。下面这个例子借助after_response钩子实现每 30 秒最多清理一次的节流式清理见 docs/examples/stores/delete_expired_after_response.pyfrom datetime import datetime, timedelta from litestar import Litestar, Request from litestar.stores.memory import MemoryStore memory_store MemoryStore() async def after_response(request: Request) - None: now datetime.utcnow() last_cleared request.app.state.get(store_last_cleared, now) if datetime.utcnow() - last_cleared timedelta(seconds30): await memory_store.delete_expired() app.state[store_last_cleared] now app Litestar(after_responseafter_response)对于FileStore还可以在应用启动时清理过期文件见 docs/examples/stores/delete_expired_on_startup.pyfrom pathlib import Path from litestar import Litestar from litestar.stores.file import FileStore file_store FileStore(Path(data)) async def on_startup() - None: await file_store.delete_expired() app Litestar(on_startup[on_startup])对于MemoryStore则无需这样的启动清理——数据就存在一个字典里每次创建新的 store 实例时都是从空开始。五、可以存储什么为什么是 bytesStore 的通用数据单元是bytesset接受bytes存储也接受字符串会自动 UTF-8 编码因此即使set时传入字符串get返回的也一定是bytes。这种限制的根源在于不同后端内存、文件、Redis、Valkey的编码、存储与反序列化能力差异巨大而 store 的设计目标是可互换interchangeable所以必须选择一个所有后端都支持的最小公共类型。bytes满足这一要求并且能表达非常广泛的数据形态。一个技术细节MemoryStore 的特殊性MemoryStore与上述规则不同——它在存储前不做任何编码因此技术上可以把任意 Python 对象存进去再取回同一个对象。但这一行为并未体现在类型签名中底层Store接口并不保证这一点也不保证MemoryStore未来始终如此。所以不要依赖这个特性。六、命名空间Namespacing安全隔离与分层管理6.1 为什么需要命名空间当一个 store 被用于多种用途时进行delete_all这类批量操作需要格外小心。例如直接对 Redis 执行FLUSHALL可能产生无法预料的后果会清掉 Redis 里所有与当前应用无关的数据。为此部分 store 提供了命名空间能力允许构建简单的 store 层级结构。这些 store 是NamespacedStore见 base.py的子类额外提供with_namespace(namespace)方法返回一个新的NamespacedStore实例。一旦创建了带命名空间的 store对它的操作只会影响它自身及其子命名空间。对RedisStore来说这意味着可以复用同一个 Redis 实例和连接同时保证不同用途的数据相互隔离。6.2 RedisStore 的默认命名空间RedisStore默认使用LITESTAR命名空间所有创建的 key 在 Redis 中都会带上LITESTAR前缀RedisStore.delete_all的实现只会删除匹配当前命名空间的 key因此是安全、无副作用的可以通过在创建实例时显式传namespaceNone来关闭这一行为。6.3 命名空间示例见 docs/examples/stores/namespacing.pyfrom litestar import Litestar from litestar.stores.redis import RedisStore root_store RedisStore.with_client() cache_store root_store.with_namespace(cache) session_store root_store.with_namespace(sessions) async def on_shutdown() - None: await cache_store.delete_all() app Litestar(on_shutdown[on_shutdown])这里三个 store 共享同一个 Redis 实例但在cache_store上调用delete_all不会影响session_store中的数据。同时这种分层结构仍保留了一键清空的能力——只要在根 storeroot_store上调用delete_all所有子命名空间的数据都会一并清除依据NamespacedStore的语义父命名空间上的批量操作应影响所有子命名空间见 base.py。七、用 StoreRegistry 统一管理存储7.1 注册表的工作机制StoreRegistry 是集中配置和管理 store 的枢纽应用内部如中间件、Litestar 内部机制以及第三方集成都可以通过它访问到统一的存储实例。它通过Litestar.stores属性在整个应用上下文中可用。它基于三个基本原则工作与 registry.py 的实现 完全对应注册可以向注册表提供一个初始的stores映射dict[str, Store]或调用register(name, store, allow_overrideFalse)注册新 store——注意默认不允许覆盖同名 store否则抛ValueError获取通过get(name)请求已注册的 store按需创建如果请求的 store 尚未注册注册表会用默认工厂default_factory以该名字创建一个新 store 并注册之后相同名字的请求都会返回同一个实例。基础示例见 docs/examples/stores/registry.pyfrom litestar import Litestar from litestar.stores.memory import MemoryStore app Litestar([], stores{memory: MemoryStore()}) memory_store app.stores.get(memory) # 返回的是前面定义的那个 store some_other_store app.stores.get(something_else) # 这个名字尚未注册会通过默认工厂新建一个实例 assert app.stores.get(something_else) is some_other_store # 后续请求返回同一个实例这种模式带来了两方面的价值store 之间的隔离以及为中间件和其他 Litestar 特性/第三方集成配置 store 的简单途径。7.2 通过注册表访问集成组件使用的 store下面的例子展示了如何通过注册表访问RateLimitMiddleware内部使用的 store见 docs/examples/stores/registry_access_integration.pyfrom litestar import Litestar from litestar.middleware.rate_limit import RateLimitConfig app Litestar(middleware[RateLimitConfig((second, 1)).middleware]) rate_limit_store app.stores.get(rate_limit)这之所以可行是因为RateLimitMiddleware内部同样通过app.stores.get(...)来请求自己的 store——也就是说集成组件与业务代码共享同一个注册表天然打通。7.3 默认工厂default factory上面按需创建的机制依赖注册表的默认工厂一个可调用对象每当请求的 store 尚未注册时被调用。它类似dict.get的default参数。默认情况下默认工厂是一个返回新MemoryStore实例的函数default_default_factory可以通过向注册表传入自定义default_factory来改变这一行为。该工厂接收请求的 store 名字str返回一个Store实例。自定义默认工厂的示例见 docs/examples/stores/registry_default_factory.pyfrom litestar import Litestar from litestar.stores.memory import MemoryStore from litestar.stores.registry import StoreRegistry memory_store MemoryStore() def default_factory(name: str) - MemoryStore: return memory_store app Litestar([], storesStoreRegistry(default_factorydefault_factory))此时每当请求一个未定义的 store注册表都会返回同一个MemoryStore实例。7.4 用注册表配置集成组件的存储这一机制同样可以用来控制各种集成的存储例如中间件。见 docs/examples/stores/registry_configure_integrations.pyfrom pathlib import Path from litestar import Litestar from litestar.middleware.session.server_side import ServerSideSessionConfig from litestar.stores.file import FileStore from litestar.stores.redis import RedisStore app Litestar( stores{ sessions: RedisStore.with_client(), response_cache: FileStore(Path(response-cache)), }, middleware[ServerSideSessionConfig().middleware], )这里的sessions和response_cache并不是魔法常量而是可配置的默认值。调整这些默认值可以轻松实现 store 复用无需更复杂的配置。见 docs/examples/stores/configure_integrations_set_names.pyfrom pathlib import Path from litestar import Litestar from litestar.config.response_cache import ResponseCacheConfig from litestar.middleware.rate_limit import RateLimitConfig from litestar.middleware.session.server_side import ServerSideSessionConfig from litestar.stores.file import FileStore from litestar.stores.redis import RedisStore app Litestar( stores{redis: RedisStore.with_client(), file: FileStore(Path(data))}, response_cache_configResponseCacheConfig(storeredis), middleware[ ServerSideSessionConfig(storefile).middleware, RateLimitConfig(rate_limit(second, 10), storeredis).middleware, ], )这样配置后限流中间件RateLimitConfig使用redisstore响应缓存ResponseCacheConfig(storeredis)使用redisstore服务端会话ServerSideSessionConfig(storefile)使用filestore。7.5 默认工厂 命名空间零样板的分层隔离把默认工厂与命名空间结合可以用极少的样板代码创建相互隔离、层级分明的存储结构。见 docs/examples/stores/registry_default_factory_namespacing.pyfrom litestar import Litestar, get from litestar.middleware.rate_limit import RateLimitConfig from litestar.middleware.session.server_side import ServerSideSessionConfig from litestar.stores.redis import RedisStore from litestar.stores.registry import StoreRegistry root_store RedisStore.with_client() get(cacheTrue, sync_to_threadFalse) def cached_handler() - str: # 这会使用 app.stores.get(response_cache) return Hello, world! app Litestar( [cached_handler], storesStoreRegistry(default_factoryroot_store.with_namespace), middleware[ RateLimitConfig((second, 1)).middleware, ServerSideSessionConfig().middleware, ], )这里的点睛之笔是把root_store.with_namespace方法本身作为默认工厂传入。由于默认工厂接收的参数name恰好就是with_namespace的命名空间参数因此无需任何额外配置每次以新名字调用app.stores.get(...)都会得到一个仅属于该名字的命名空间所有命名空间共享底层的同一个 Redis 实例限流、会话、响应缓存cached_handler上cacheTrue触发的response_cachestore各自隔离、互不干扰。八、Store 生命周期与资源释放需要特别留意store 在应用关闭时可能不会被自动关闭。这一点尤其适用于RedisStore——如果不是通过类方法RedisStore.with_client()创建、而是自己传入 Redis 实例那么你需要自己负责关闭该 Redis 实例。反过来通过with_client()创建的 store 会自行管理客户端生命周期应用关闭时由 Litestar 负责清理。对于MemoryStore与FileStore则不存在连接资源需要释放只需按第四节所述定期清理过期数据即可。九、总结如何选择与规划你的存储方案综合以上内容可以给出如下选型建议需求推荐 Store关键理由单进程内缓存、追求最低开销MemoryStore默认实现零持久化开销数据量大、需持久化、易备份FileStore落盘存储支持命名空间多 worker、分布式、需 TTL 等高级特性RedisStore/ValkeyStore原生过期、原子性、SCAN模式删除、支持命名空间多种用途并存、需要隔离与统一管理StoreRegistry 命名空间一个注册表、一个连接、按名字隔离配套的可运行示例都收录在 docs/examples/stores 目录下get_set、expiry、expiry_renew_on_get、delete_expired_after_response、delete_expired_on_startup、namespacing、registry、registry_access_integration、registry_default_factory、registry_configure_integrations、configure_integrations_set_names、registry_default_factory_namespacing共 12 个文件对应单元测试可参考 tests/unit/test_stores.py 与 tests/examples/test_stores.py。核心接口定义与注册表实现分别位于 litestar/stores/base.py 和 litestar/stores/registry.py是深入理解 store 语义的第一手资料。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考