使用方法：
pip install kxy-framework
from kxy.framework.context import *
数据权限过滤
框架会在 BaseDal 的统一查询、更新、删除入口自动追加数据权限过滤条件。使用方一般只需要完成两件事：
- 在模型类上声明哪些字段参与数据权限。
- 在请求入口写入当前登录用户的数据权限上下文。
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等值过滤。
Id 游标分页
大数据量深分页场景可使用 page_by_id_query，通过主键索引定位下一页，避免 OFFSET 扫描：
# 第一页，默认按 Id 降序
items = await customer_dal.page_by_id_query(filters, None, 20)
# 将本页末项 Id 作为下一页游标
last_id = items[-1].Id if items else None
next_items = await customer_dal.page_by_id_query(filters, last_id, 20)
升序遍历时传入 descending=False：
items = await customer_dal.page_by_id_query(
filters,
last_id,
20,
descending=False,
)
该函数不执行总数统计，只返回实体列表，且 page_size 必须大于 0。需要总数或按任意字段排序时，继续使用现有 paginate_query。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')
1 maintainer
Release files for kxy-framework 1.4.18
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| kxy_framework-1.4.18.tar.gz | 44.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| kxy_framework-1.4.18-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 119.7 kB
Release files / kxy_framework-1.4.18.tar.gz
| Download URL | kxy_framework-1.4.18.tar.gz |
|---|---|
| Size | 44.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
23191385b18bdde993d794b0fa9fcbff78327d5b13e073edea1119251f61c98e
|
|
BLAKE2b-256 checksum How to use checksums |
aa2415a0e91ec9626247b8fbe1ccc4e11a21e52df49d1557fcd732108b5ec818
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.10.12
|
Release files / kxy_framework-1.4.18-py3-none-any.whl
| Download URL | kxy_framework-1.4.18-py3-none-any.whl |
|---|---|
| Size | 75.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
8cbecde003cd70aaa9416dee115aa4315473237ec3d59517b81842716d29a93e
|
|
BLAKE2b-256 checksum How to use checksums |
0f810385d24b3fa43c12b1a659ed6832000857ec2e7d4ab1f880dbde58c95b56
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.10.12
|
