ށidZddlmZddlZddlmZddlmZddl m Z ddl m Z e e ZdZdZGd d ZdS) uWRedis async client wrapper for permission caching and pub/sub. Redis 异步客户端封装模块。 主要职责： 1. 用户权限缓存 —— 避免每次请求都重新解析 JWT 构建 ACL 令牌 2. 任务消息发布/订阅 —— 用于 Celery 之外的轻量级任务通知 3. 健康检查 —— 供启动探针和就绪探针使用 ) annotationsN)Any)settings) get_loggerz zm:perm:user:zzm:tasksceZdZdZd"dZed#dZed$d Zd%d Z d&dZ ddd'dZ d(dZ d)dZ d*dZd+d,dZd-d!ZdS). RedisClientuThin async wrapper around ``redis.asyncio.Redis``. 对 redis.asyncio 的薄封装，提供权限缓存、Pub/Sub 消息和健康检查接口。 clientaioredis.RedisreturnNonec||_dSN_r)selfr s 9D:\work\zm-rag\backend\app\infrastructure\redis_client.py__init__zRedisClient.__init__!s  'RedisClient'cttdd}tjtjd|}||S)uBuild a client from the global application settings. 连接池大小优先从 settings 读取，未配置则使用默认值 20。 Pool size is read from settings if available, otherwise defaults to 20. redis_max_connectionsT)decode_responsesmax_connections)getattrraioredisfrom_url redis_url)clsrr s r from_settingszRedisClient.from_settings$sI"(,CRHH"  !+    s6{{rc|jSrrrs rrawzRedisClient.raw5s wrcHK|jd{VdSr)racloser"s rclosezRedisClient.close9s0gnnruser_idstrdict[str, Any] | NonecKt|}|j|d{V}|dS tj|S#tjt f$r td|YdSwxYw)zRetrieve cached permission data for *user_id*. Returns ``None`` on cache miss. The cached value is expected to be a dict with an ``acl_tokens`` key (list of prefixed IDs). Nredis_bad_perm_cache)r') _PERM_PREFIXrgetjsonloadsJSONDecodeError TypeErrorloggerwarning)rr'keyr#s rget_user_permissionsz RedisClient.get_user_permissions>s (w((GKK$$$$$$$$ ;4 :c?? "$i0    NN17N C C C44 sA2A=<A=N)ttl permissionsdict[str, Any]r6 int | NonecKt|}|p tj}|j|t j|d|d{VdS)zECache *permissions* for *user_id* with a TTL (default from settings).F ensure_ascii)exN)r,rredis_permissions_ttlrsetr.dumps)rr'r7r6r4s rset_user_permissionsz RedisClient.set_user_permissionsNsf(w((3X3gkk#tz+EJJJskSSSSSSSSSSSrc\K|jt|d{VdS)z0Remove the cached permissions for a single user.N)rdeleter,)rr's rinvalidate_user_permissionsz'RedisClient.invalidate_user_permissionsZs=gnn 7g7788888888888r task_payloadcK|jttj|dd{Vt dt|dS)z3Publish a task message to the shared Redis channel.Fr;Nredis_task_published) payload_keys) rpublish _TASK_CHANNELr.r@r2debuglistkeys)rrEs r publish_taskzRedisClient.publish_task`sogoomTZ SX-Y-Y-YZZZZZZZZZ +$|?P?P?R?R:S:S TTTTTraioredis.client.PubSubc|K|j}|td{V|S)u~返回已订阅任务频道的 Pub/Sub 对象。 Return an async pub/sub object subscribed to the task channel. Usage:: pubsub = await redis_client.subscribe_tasks() async for message in pubsub.listen(): if message["type"] == "message": data = json.loads(message["data"]) ... N)rpubsub subscriberJ)rrQs rsubscribe_taskszRedisClient.subscribe_tasksesD!!}--------- rnametimeoutintc:|j||S)u返回一个 Redis 分布式锁（异步上下文管理器），用于保护并发写入关键区。 Return a Redis distributed lock (async context manager) for protecting critical sections against concurrent writes. 调用方式: async with redis_client.lock("my_lock", timeout=30): ... 注意: 这是同步方法，底层 raw.lock() 本身返回异步上下文管理器，无需 await。 )rV)rlock)rrUrVs rrYzRedisClient.lockwsw||D'|222rboolchK |jd{VS#t$rYdSwxYw)z1Return ``True`` if the Redis server is reachable.NF)rping Exceptionr"s rr\zRedisClient.pingsO '''''' '   55 s # 11)r r r r )r r)r r )r r )r'r(r r))r'r(r7r8r6r9r r )r'r(r r )rEr8r r )r rO)rT)rUr(rVrW)r rZ)__name__ __module__ __qualname____doc__r classmethodr propertyr#r&r5rArDrNrSrYr\rrrrs0 [ X * T T T T T T9999 UUUU $ 3 3 3 3 3rr)ra __future__rr.typingr redis.asyncioasyncior app.configrapp.utils.loggerrr^r2r,rJrrdrrrks#"""""  '''''' H    nnnnnnnnnnr