修复rbac_compute_pricing导入错误
- 修复rbac模块导入路径:从 ..rbac 改为 ...rbac - 修复require_role不存在的问题:使用role_allowed函数创建check_role辅助函数 - 替换所有require_role调用为check_role
This commit is contained in:
@@ -19,7 +19,8 @@ from pydantic import BaseModel
|
||||
from sqlalchemy import select, and_
|
||||
|
||||
from ..dependencies import get_db, get_current_user
|
||||
from ..rbac import write_audit, require_role
|
||||
from ...rbac import write_audit
|
||||
from ...domain.rules import role_allowed
|
||||
from ...infrastructure.models import (
|
||||
ParkTenant, ParkCompany, CompanyMember,
|
||||
TenantDiscount, TenantUserDiscount,
|
||||
@@ -43,6 +44,12 @@ def new_id() -> str:
|
||||
return uuid.uuid4().hex[:24]
|
||||
|
||||
|
||||
def check_role(user: dict, roles: list[str]) -> None:
|
||||
"""检查用户角色,如果不允许则抛出403异常。"""
|
||||
if not role_allowed(user, *roles):
|
||||
raise HTTPException(status_code=403, detail="Forbidden: insufficient role")
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
# 运营端:载体折扣管理
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
@@ -60,7 +67,7 @@ async def admin_tenant_discounts(
|
||||
db: Database = Depends(get_db),
|
||||
user: dict = Depends(get_current_user),
|
||||
):
|
||||
require_role(user, ["operator"])
|
||||
check_role(user, ["operator"])
|
||||
result = await db.session.execute(select(TenantDiscount).order_by(TenantDiscount.created_at.desc()))
|
||||
items = []
|
||||
for td in result.scalars().all():
|
||||
@@ -86,7 +93,7 @@ async def admin_set_tenant_discount(
|
||||
db: Database = Depends(get_db),
|
||||
user: dict = Depends(get_current_user),
|
||||
):
|
||||
require_role(user, ["operator"])
|
||||
check_role(user, ["operator"])
|
||||
body.discount = max(0, min(100, int(body.discount)))
|
||||
|
||||
# 检查载体是否存在
|
||||
@@ -139,7 +146,7 @@ class StandardPriceBody(BaseModel):
|
||||
async def admin_standard_prices(
|
||||
user: dict = Depends(get_current_user),
|
||||
):
|
||||
require_role(user, ["operator"])
|
||||
check_role(user, ["operator"])
|
||||
items = [{"model": k, "price_per_1k": v} for k, v in STANDARD_PRICES.items()]
|
||||
return {"items": items}
|
||||
|
||||
@@ -152,7 +159,7 @@ async def admin_update_standard_price(
|
||||
db: Database = Depends(get_db),
|
||||
user: dict = Depends(get_current_user),
|
||||
):
|
||||
require_role(user, ["operator"])
|
||||
check_role(user, ["operator"])
|
||||
STANDARD_PRICES[model] = max(1, int(body.price_per_1k))
|
||||
await write_audit(db, action="compute.standard_price_update", resource="standard_price",
|
||||
resource_id=model, detail=f"price={body.price_per_1k}",
|
||||
|
||||
Reference in New Issue
Block a user