feat: 新增 Redis 查询缓存加速

新建 cache.py 缓存工具模块,封装 Redis 读写和容错降级。
订单详情/列表、客户列表、定价规则读取走缓存,写操作后精确清除。
Redis 不可用时自动降级查 DB,不影响正常业务。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
taiyi 2026-05-31 10:37:51 +08:00
parent 134a3824d3
commit 0adccb60d2
5 changed files with 169 additions and 3 deletions

View File

@ -15,6 +15,7 @@ from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from backend.app.api.deps import get_current_user, require_permissions, require_roles
from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set
from backend.app.core.exceptions import AppException
from backend.app.core.error_codes import ErrorCode
from backend.app.db import get_db_session
@ -88,13 +89,19 @@ def get_pricing_rule(
返回值定价规则详情信息
权限要求管理员admin或经理manager
"""
cache_key = f"pricing:rule:{product_id}"
cached = cache_get(cache_key)
if cached is not None:
return success_payload(cached)
rule = session.query(ProductPricingRule).filter(
ProductPricingRule.product_id == product_id,
ProductPricingRule.deleted == 0,
).first()
if not rule:
raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
return success_payload(_rule_to_dict(rule))
rule_data = _rule_to_dict(rule)
cache_set(cache_key, rule_data, ttl=300)
return success_payload(rule_data)
@router.post("/pricing-rules")
@ -132,6 +139,7 @@ def create_pricing_rule(
)
session.add(rule)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(rule)
return success_payload(_rule_to_dict(rule))
@ -161,6 +169,7 @@ def update_pricing_rule(
for k, v in data.items():
setattr(rule, k, v)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(rule)
return success_payload(_rule_to_dict(rule))
@ -187,6 +196,7 @@ def delete_pricing_rule(
raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
rule.deleted = 1
session.commit()
cache_delete_pattern("pricing:*")
return success_payload({"deleted": True})
@ -230,6 +240,7 @@ def create_supplier_cost(
cost = SupplierProductCost(**payload.model_dump())
session.add(cost)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(cost)
return success_payload(_cost_to_dict(cost))
@ -258,6 +269,7 @@ def update_supplier_cost(
for k, v in data.items():
setattr(cost, k, v)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(cost)
return success_payload(_cost_to_dict(cost))
@ -283,6 +295,7 @@ def delete_supplier_cost(
raise AppException(code=ErrorCode.NOT_FOUND, message="供应商成本不存在", status_code=404)
cost.deleted = 1
session.commit()
cache_delete_pattern("pricing:*")
return success_payload({"deleted": True})
@ -326,6 +339,7 @@ def create_price_tier(
tier = ProductPriceTier(**payload.model_dump())
session.add(tier)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(tier)
return success_payload(_tier_to_dict(tier))
@ -354,6 +368,7 @@ def update_price_tier(
for k, v in data.items():
setattr(tier, k, v)
session.commit()
cache_delete_pattern("pricing:*")
session.refresh(tier)
return success_payload(_tier_to_dict(tier))
@ -379,6 +394,7 @@ def delete_price_tier(
raise AppException(code=ErrorCode.NOT_FOUND, message="价格层级不存在", status_code=404)
tier.deleted = 1
session.commit()
cache_delete_pattern("pricing:*")
return success_payload({"deleted": True})

105
backend/app/core/cache.py Normal file
View File

@ -0,0 +1,105 @@
"""Redis 缓存工具模块。
提供 Redis 连接管理和缓存读写操作所有操作在 Redis 不可用时静默降级
OrderServiceCustomerServicepricing 路由等调用
"""
import hashlib
import json
import logging
import redis
from backend.app.core.config import get_settings
logger = logging.getLogger(__name__)
_redis_client: redis.Redis | None = None
def get_redis() -> redis.Redis | None:
"""获取 Redis 连接单例,连接失败时返回 None。"""
global _redis_client
if _redis_client is not None:
return _redis_client
try:
settings = get_settings()
_redis_client = redis.Redis(
host=settings.redis_host,
port=settings.redis_port,
db=settings.redis_db,
password=settings.redis_password or None,
decode_responses=True,
socket_connect_timeout=2,
socket_timeout=2,
)
_redis_client.ping()
logger.info("Redis 连接成功: %s:%s/%s", settings.redis_host, settings.redis_port, settings.redis_db)
return _redis_client
except Exception as exc:
logger.warning("Redis 连接失败,降级为无缓存模式: %s", exc)
_redis_client = None
return None
def cache_get(key: str):
"""读取缓存,返回反序列化后的 dict/list未命中或异常返回 None。"""
try:
r = get_redis()
if r is None:
return None
raw = r.get(key)
if raw is None:
return None
return json.loads(raw)
except Exception:
return None
def cache_set(key: str, data, ttl: int = 60) -> None:
"""写入缓存JSON 序列化后存储并设置 TTL。异常时静默跳过。"""
try:
r = get_redis()
if r is None:
return
r.set(key, json.dumps(data, ensure_ascii=False, default=str), ex=ttl)
except Exception:
pass
def cache_delete(key: str) -> None:
"""精确删除单个缓存 key。"""
try:
r = get_redis()
if r is not None:
r.delete(key)
except Exception:
pass
def cache_delete_pattern(pattern: str) -> None:
"""按模式批量删除缓存SCAN + DELETE"order:list:*""""
try:
r = get_redis()
if r is None:
return
cursor = 0
while True:
cursor, keys = r.scan(cursor=cursor, match=pattern, count=100)
if keys:
r.delete(*keys)
if cursor == 0:
break
except Exception:
pass
def make_cache_key(prefix: str, **params) -> str:
"""生成带参数哈希的缓存 key。
: make_cache_key("order:list", user_id=1, status="draft")
"order:list:a1b2c3d4"
"""
raw = json.dumps(params, sort_keys=True, ensure_ascii=False, default=str)
suffix = hashlib.md5(raw.encode()).hexdigest()[:8]
return f"{prefix}:{suffix}"

View File

@ -47,6 +47,10 @@ class Settings(BaseSettings):
mysql_user: str = Field(default="order_user", alias="MYSQL_USER")
mysql_password: str = Field(default="password", alias="MYSQL_PASSWORD")
sql_echo: bool = Field(default=False, alias="SQL_ECHO")
redis_host: str = Field(default="127.0.0.1", alias="REDIS_HOST")
redis_port: int = Field(default=6379, alias="REDIS_PORT")
redis_db: int = Field(default=0, alias="REDIS_DB")
redis_password: str = Field(default="", alias="REDIS_PASSWORD")
aliyun_oss_bucket: str = Field(default="bucket-name", alias="ALIYUN_OSS_BUCKET")
aliyun_oss_endpoint: str = Field(default="oss-cn-hangzhou.aliyuncs.com", alias="ALIYUN_OSS_ENDPOINT")
aliyun_oss_public_base_url: str = Field(default="https://cdn.example.com", alias="ALIYUN_OSS_PUBLIC_BASE_URL")

View File

@ -11,6 +11,7 @@ from pathlib import Path
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set, make_cache_key
from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.repositories.customer_repository import CustomerRepository
@ -44,9 +45,14 @@ class CustomerService:
"""
if session is None:
raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
# 读缓存
cache_key = make_cache_key("customer:list", **(filters or {}))
cached = cache_get(cache_key)
if cached is not None:
return cached
try:
customers = self.repository.list_customers_by_filters(session, filters or {})
return {
result = {
"total": len(customers),
"page_no": 1,
"page_size": 20,
@ -67,6 +73,8 @@ class CustomerService:
for item in customers
],
}
cache_set(cache_key, result, ttl=60)
return result
except SQLAlchemyError as exc:
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc
@ -136,6 +144,7 @@ class CustomerService:
},
)
session.commit()
cache_delete_pattern("customer:list:*")
return {
"customer_id": customer.id,
"customer_name": customer.customer_name,
@ -212,6 +221,7 @@ class CustomerService:
},
)
session.commit()
cache_delete_pattern("customer:list:*")
return {"customer_id": customer.id, "customer_name": customer.customer_name, "mobile": customer.mobile}
except AppException:
session.rollback()
@ -314,6 +324,7 @@ class CustomerService:
},
)
session.commit()
cache_delete_pattern("customer:list:*")
return result
except AppException:
session.rollback()

View File

@ -12,6 +12,7 @@ import json
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from backend.app.core.cache import cache_delete, cache_delete_pattern, cache_get, cache_set, make_cache_key
from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.repositories.customer_repository import CustomerRepository
@ -65,11 +66,16 @@ class OrderService:
被调用路由: orders.py - GET /orders, salesman.py - GET /salesman/orders
"""
if session is not None:
# 读缓存
cache_key = make_cache_key("order:list", user_id=(current_user or {}).get("user_id"), **(filters or {}))
cached = cache_get(cache_key)
if cached is not None:
return cached
try:
result = self.order_repository.list_orders_by_filters(session, filters or {})
orders = result["list"]
role_code = (current_user or {}).get("role_code")
return [
rows = [
self._filter_order_row(
{
"order_id": item.id,
@ -88,6 +94,9 @@ class OrderService:
)
for item in orders
]
# 写缓存
cache_set(cache_key, rows, ttl=30)
return rows
except SQLAlchemyError as exc:
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
return []
@ -114,6 +123,11 @@ class OrderService:
被调用路由: orders.py - GET /orders/{order_id}
"""
if session is not None:
# 读缓存
cache_key = f"order:detail:{order_id}"
cached = cache_get(cache_key)
if cached is not None:
return self._filter_order_detail_by_role(cached, current_user)
try:
order = self.order_repository.get_order(session, order_id)
if order is None:
@ -213,6 +227,8 @@ class OrderService:
"profit_alert_threshold": float(self._get_config_value(session, "profit_alert_threshold", "0")),
"remark": order.remark,
}
# 写缓存(缓存完整结果,角色过滤在读取后执行)
cache_set(cache_key, result, ttl=60)
return self._filter_order_detail_by_role(result, current_user)
except SQLAlchemyError as exc:
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
@ -325,6 +341,7 @@ class OrderService:
},
)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_no": order.order_no,
@ -388,6 +405,7 @@ class OrderService:
)
self._notify_status_change(session, order, before_status, order.order_status, current_user.get("user_id") if current_user else None)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -487,6 +505,7 @@ class OrderService:
)
self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -567,6 +586,7 @@ class OrderService:
)
self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None, {"approve_opinion": payload.get("approve_opinion")})
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -655,6 +675,7 @@ class OrderService:
)
self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -782,6 +803,7 @@ class OrderService:
},
)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -864,6 +886,7 @@ class OrderService:
)
self._notify_status_change(session, order, before_status, order.order_status, current_user.get("user_id") if current_user else None)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_status": order.order_status,
@ -988,6 +1011,7 @@ class OrderService:
},
)
session.commit()
self._invalidate_order_cache(order.id)
return {
"order_id": order.id,
"order_no": order.order_no,
@ -1199,6 +1223,12 @@ class OrderService:
return filtered
return order
def _invalidate_order_cache(self, order_id: int | None = None) -> None:
"""清除订单相关缓存。"""
if order_id:
cache_delete(f"order:detail:{order_id}")
cache_delete_pattern("order:list:*")
def _filter_order_row(self, row: dict, role_code: str | None) -> dict:
"""根据用户角色过滤订单列表行中的敏感字段。