kxy-framework 1.4.16

封装了ORM框架

使用方法：

pip install kxy-framework

from kxy.framework.context import *

数据权限过滤

框架会在 BaseDal 的统一查询、更新、删除入口自动追加数据权限过滤条件。使用方一般只需要完成两件事：

1. 在模型类上声明哪些字段参与数据权限。
2. 在请求入口写入当前登录用户的数据权限上下文。

1. 模型声明

推荐使用 DataPermission 一次性声明租户、部门、本人字段：

from kxy.framework.filter import DataPermission

@DataPermission(
    tenant_field="TenantId",
    dept_field="DepId",
    user_field="CreateUser",
)
class Customer(BaseEntity, Modal):
    __tablename__ = "customer"

也兼容旧写法：

from kxy.framework.filter import FilterTenant, FilterDepartment, FilterUser

@FilterTenant("TenantId")
@FilterDepartment("DepId")
@FilterUser("CreateUser")
class Customer(BaseEntity, Modal):
    __tablename__ = "customer"

2. 请求入口设置上下文

基础库不负责查询角色、组织和部门树。业务系统应在登录后或请求中间件中计算好当前用户最终可见范围，再写入上下文：

from kxy.framework.context import user_id, current_tenant_id, dep_id
from kxy.framework.data_permission import (
    DataScope,
    set_data_permission_context,
)

user_id.set(login_user["id"])
current_tenant_id.set(login_user["tenant_id"])
dep_id.set(login_user["dept_id"])

# 示例：可看本部门及子部门，同时可看本人创建的数据
set_data_permission_context(
    scope=DataScope.DEPT_AND_CHILD,
    dept_ids=login_user["visible_dept_ids"],
    child_dept_ids=login_user["child_dept_ids"],
    allow_self=True,
)

3. 业务查询无需手动拼权限条件

items, total = await customer_dal.paginate_query(
    filters,
    Customer.CreateDate.desc(),
    page_index,
    page_size,
)

如果模型声明了 tenant_field="TenantId"、dept_field="DepId"、user_field="CreateUser"，最终效果类似：

WHERE 业务条件
  AND TenantId = 当前租户
  AND (DepId IN (可见部门) OR CreateUser = 当前用户)

4. 数据范围说明

DataScope.ALL             # 全部数据，不追加部门/本人过滤
DataScope.DEPT_CUSTOM     # 指定部门，使用 data_scope_dept_ids
DataScope.DEPT_ONLY       # 当前部门，使用 dep_id
DataScope.DEPT_AND_CHILD  # 当前部门及子部门，使用 dep_id + child_dep_ids + data_scope_dept_ids
DataScope.SELF            # 仅本人，使用 user_field == user_id

多角色并集建议由业务系统先计算成最终结果，再写入 data_scope_dept_ids、self_data_permission 等上下文。

如果部门配置为 *，表示全部部门，框架不会追加部门/本人数据权限条件，只保留租户等其它过滤：

set_data_permission_context(
    scope=DataScope.DEPT_CUSTOM,
    dept_ids=["*"],
)

5. 临时忽略数据权限

忽略全部过滤：

from kxy.framework.data_permission import data_permission_ignore

async with data_permission_ignore():
    data = await customer_dal.QueryWhere([Customer.Id == id])

只忽略部门/本人过滤，保留租户过滤：

async with data_permission_ignore("dept", "self"):
    data = await customer_dal.QueryWhere([Customer.Id == id])

也可以用于方法装饰器：

from kxy.framework.data_permission import ignore_data_permission

@ignore_data_permission("dept", "self")
async def check_unique(self, name):
    return await self.QueryOne([Customer.Name == name])

6. 注意事项

• BaseDal.Get、QueryWhere、QueryOne、QueryCount、paginate_query、UpdateFields、DeleteWhere 等会自动追加过滤。
• ExecSql(sql: str) 是裸 SQL，框架无法安全改写，不会自动注入数据权限；使用时需要业务方自行拼接权限条件。
• 未设置 data_scope 时，框架保持旧版行为：FilterDepartment 按 dep_id 等值过滤，FilterUser 按 user_id 等值过滤。

日志

from kxy.framework.slogger import create_logger

logger,handler = create_logger(logging.DEBUG,'appName','production',filename='log/app',file_type='log',backupCount=5,maxBytes=10485760,mutiple_process=False)

# mutiple_process 是否多进程，多进程需要生成不同的文件

logger.info('hello world',extra={"logCategory":'http'})

日志进阶用法，设置traceId：

from kxy.framework.slogger import user_id,user_info,new_trace
req_trace_id = request.headers.get("X-Trace-ID") or str(uuid4())
new_trace(req_trace_id)
user_id.set('login user id')
# user_info用于存储用户自定义的信息，方便在其他地方取用
user_info.set({'name':'login user name','email':'login user email'})

日志自定义打印格式：

class JsonFormatter(logging.Formatter):
    def format(self, record):
        # 构建日志记录的字典
        log_record = {
            "appName":appName,
            "serverAddr":os.environ.get('IP',localIp),
            "cluster": env,
            "levelname": record.levelname,
            "filename": record.filename,
            "lineno": record.lineno,
            "traceId": record.trace_id,
            "sessionId": record.session_id,
            "userId":record.userid,
            "seqId": record.seq,
            "message": record.getMessage(),
            "CreateTime": self.formatTime(record, self.datefmt),
            "createdOn": int(time.time() * 1000)  # 添加 Unix 时间戳
        }
        # 将字典转换为 JSON 字符串
        return json.dumps(log_record, ensure_ascii=False)

from simple_util.slogger import create_logger

logger,handler = create_logger(logging.DEBUG,'appName','production',filname='log/app',file_type='log',backupCount=5,maxBytes=10485760)
handler.setFormatter(JsonFormatter())
logger.info('hello world')