feat(F009): implement third-party org sync module (Feishu + GenericAPI)

Add org_sync DDD module with Provider + Reconciler architecture for
syncing departments and members from third-party platforms (Feishu,
Generic REST API) into BiSheng. Includes 9 API endpoints, Celery async
execution, Beat cron scheduling, Fernet-encrypted credential storage,
dual-lock concurrency protection, and OpenFGA tuple auto-maintenance
via DepartmentChangeHandler.

- OrgSyncConfig + OrgSyncLog ORM models with DAO (T-01)
- Alembic migration for new tables + User source/external_id extension (T-02, T-04)
- Error codes 22000-22009, module 220 (T-03)
- OrgSyncProvider ABC with lazy-import factory (T-05)
- FeishuProvider: BFS dept traversal, token cache, 429 backoff (T-06)
- GenericAPIProvider: configurable endpoints + field mapping (T-07)
- WeComProvider + DingTalkProvider stubs (T-08)
- Reconciler: pure-logic dept/member diff engine with topo sort (T-09, T-10)
- OrgSyncService: 16-step orchestrator with partial failure handling (T-11)
- Celery tasks + Beat schedule for cron-based sync (T-12)
- Request/response DTOs with sensitive field masking (T-13)
- Config CRUD + execution API endpoints (T-14, T-15)
- 30 tests passing (18 reconciler + 12 API schema) (T-16, T-17)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
LineWalker
2026-04-13 17:13:14 +08:00
parent 0935ddc797
commit 1eb5d58e98
35 changed files with 4902 additions and 25 deletions
File diff suppressed because it is too large Load Diff
+434 -4
View File
@@ -9,10 +9,440 @@
| 步骤 | 状态 | 备注 |
|------|------|------|
| spec.md | 🔲 草稿 | P2 延后,用户确认后改为 ✅ 已评审 |
| tasks.md | 🔲 草稿 | 拆解完成后改为 ✅ 已拆解 |
| 实现 | 🔲 未开始 | 0 / N 完成 |
| spec.md | ✅ 已评审 | 2026-04-13 审查通过(2 个 low 已修复) |
| tasks.md | ✅ 已拆解 | 2026-04-13 审查通过(Round 2 LGTM |
| 实现 | ✅ 已完成 | 17 / 17 完成2026-04-13 |
---
_P2 延后,待 spec.md 评审通过后拆解任务_
## 任务列表
### Phase 1: Foundation
#### T-01: OrgSyncConfig + OrgSyncLog ORM 模型与 DAO
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/models/__init__.py`
- 新建: `src/backend/bisheng/org_sync/domain/models/org_sync.py`
**内容**:
- OrgSyncConfig ORMspec §5 定义),含 tenant_id、所有字段、UniqueConstraint
- OrgSyncLog ORMspec §5 定义),含 tenant_id、统计字段、error_details JSON
- OrgSyncConfigDaoacreate / aget_by_id / aget_list / aupdate / aset_sync_status / aget_active_cron_configs
- OrgSyncLogDaoacreate / aupdate / aget_by_config(分页)
- 加密/解密辅助函数:encrypt_auth_config / decrypt_auth_configFernetAD-02
**覆盖 AC**: AC-01ORM 结构), AC-02(唯一约束)
**验证**: 单元测试——SQLite in-memory 验证 CRUD + 唯一约束 + 加解密
---
#### T-02: 数据库迁移脚本
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/core/database/alembic/versions/v2_5_0_f009_org_sync.py`
**内容**:
- CREATE TABLE org_sync_config(所有列 + 索引 + UK
- CREATE TABLE org_sync_log(所有列 + 索引)
- ALTER TABLE user ADD COLUMN source VARCHAR(32) NOT NULL DEFAULT 'local'
- ALTER TABLE user ADD COLUMN external_id VARCHAR(128) NULL
- CREATE UNIQUE INDEX uk_user_source_external_id ON user(source, external_id)
**依赖**: T-01(模型定义确定后写迁移)
**验证**: 在 MySQL 上执行 upgrade/downgrade 无报错
---
#### T-03: 错误码模块 220
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/common/errcode/org_sync.py`
**内容**:
- 10 个错误码类(22000~22009),继承 BaseErrorCode
- 类名遵循 `OrgSync{Error}Error` 命名
**覆盖 AC**: AC-02, AC-08, AC-10, AC-12, AC-14, AC-15, AC-33
**验证**: import 无报错,Code/Msg 属性正确
---
#### T-04: User 模型扩展 + UserDao 新方法
- [x] 完成
**文件**:
- 修改: `src/backend/bisheng/user/domain/models/user.py`
**内容**:
- User 类新增 source 字段(default='local'+ external_id 字段(nullable
- 新增 __table_args__ 中的 UniqueConstraint('source', 'external_id')
- UserDao 新增 aget_by_source_external_id(source, external_id)
- UserDao 新增 aget_by_source(source, tenant_id)(查某来源全部用户)
**覆盖 AC**: AC-22User 匹配依赖 source+external_id
**验证**: 单元测试——新字段默认值正确,DAO 方法可查询
---
#### T-05: Provider 抽象基类 + 远程 DTO
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/providers/__init__.py`
- 新建: `src/backend/bisheng/org_sync/domain/providers/base.py`
- 新建: `src/backend/bisheng/org_sync/domain/schemas/__init__.py`
- 新建: `src/backend/bisheng/org_sync/domain/schemas/remote_dto.py`
**内容**:
- OrgSyncProvider ABCauthenticate / fetch_departments / fetch_members / test_connection
- get_provider(provider, auth_config) 工厂方法
- RemoteDepartmentDTOexternal_id, name, parent_external_id, sort_order
- RemoteMemberDTOexternal_id, name, email, phone, primary_dept_external_id, secondary_dept_external_ids, status
**覆盖 AC**: AC-12Provider 未实现时 raise OrgSyncProviderError
**验证**: import 无报错,工厂方法对未知 provider 抛出正确错误
---
### Phase 2: Providers
#### T-06: FeishuProvider 完整实现
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/providers/feishu.py`
**内容**:
- authenticate()POST /auth/v3/tenant_access_token/internal → 获取 tenant_access_token
- fetch_departments()GET /contact/v3/departments BFS 遍历 + page_token 分页,返回 RemoteDepartmentDTO 列表
- fetch_members()GET /contact/v3/users?department_id=X 逐部门拉取 + page_token 分页
- test_connection()authenticate + GET /contact/v3/departments/0 获取根部门元数据
- httpx.AsyncClientSemaphore(5) 并发控制,429 指数退避重试(1s/2s/4s
- token 缓存(实例属性,2h 有效期)
**覆盖 AC**: AC-09, AC-10, AC-11
**验证**: mock httpx 响应的单元测试——认证成功/失败、部门分页、人员拉取、429 重试
---
#### T-07: GenericAPIProvider 完整实现
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/providers/generic_api.py`
**内容**:
- authenticate():根据 auth_type 验证连接(api_key → 带 header/query 参数请求测试端点;password → 基本认证)
- fetch_departments()GET departments_url,通过 field_mapping 转换为 RemoteDepartmentDTO
- fetch_members()GET members_url,通过 field_mapping 转换为 RemoteMemberDTO
- test_connection()authenticate + 基本 GET 验证
- 字段映射逻辑:从 auth_config.field_mapping 读取,有默认映射
**覆盖 AC**: AC-09, AC-11
**验证**: mock httpx 响应的单元测试——标准格式 + 自定义映射 + 格式错误处理
---
#### T-08: WeComProvider + DingTalkProvider stub
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/providers/wecom.py`
- 新建: `src/backend/bisheng/org_sync/domain/providers/dingtalk.py`
**内容**:
- 继承 OrgSyncProvider4 个方法均 raise OrgSyncProviderError(msg="WeChat Work provider not implemented")
- 类文档注释说明 API 契约(留给后续实现者参考)
**覆盖 AC**: AC-12
**验证**: 调用任何方法均抛出 OrgSyncProviderError
---
### Phase 3: Sync Engine
#### T-09: Reconciler — 部门差异引擎
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/services/__init__.py`
- 新建: `src/backend/bisheng/org_sync/domain/services/reconciler.py`
**内容**:
- reconcile_departments(remote_depts, local_depts, source) → list[DeptOperation]
- DeptOperation 数据类:CreateDept / UpdateDept / MoveDept / ArchiveDept
- 逻辑:构建 remote_map + local_map → 创建/更新/移动/归档判定 → 拓扑排序(创建父先子后,归档子先父后)
- 本地冲突处理:source='local' 的部门匹配远程 → 强制覆盖 + 改 sourceAC-18
- 循环引用检测:拓扑排序发现环 → 记录错误跳过
**覆盖 AC**: AC-16, AC-17, AC-18, AC-19, AC-20, AC-21
**验证**: 纯逻辑单元测试(无 IO),覆盖全部 6 种部门场景 + 循环引用 + 空输入
---
#### T-10: Reconciler — 人员差异引擎
- [x] 完成
**文件**:
- 修改: `src/backend/bisheng/org_sync/domain/services/reconciler.py`
**内容**:
- reconcile_members(remote_members, local_users, local_user_depts, ext_to_local_dept, source) → list[MemberOperation]
- MemberOperation 数据类:CreateMember / UpdateMember / TransferMember / DisableMember / ReactivateMember
- 逻辑:构建 remote_map + local_map → 创建/更新/转岗/禁用/重新激活判定
- 本地冲突处理:source='local' 的用户匹配远程 → 强制覆盖 + 改 source
**覆盖 AC**: AC-22, AC-24, AC-25, AC-26, AC-27, AC-28
**验证**: 纯逻辑单元测试(无 IO),覆盖全部 6 种人员场景 + 本地冲突 + 空输入
---
#### T-11: OrgSyncService — 同步编排器
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/services/org_sync_service.py`
**内容**:
- execute_sync(config_id, trigger_type, trigger_user):主编排流程(spec §7 的 16 步)
- _apply_dept_ops(ops, config, stats):执行部门操作(直接 DAO + DepartmentChangeHandlerAD-11
- _apply_member_ops(ops, config, stats, dept_map):执行人员操作(User 创建/更新/禁用 + UserDepartment + OpenFGA
- 互斥锁获取/释放:DB sync_status 原子 UPDATE + Redis 分布式锁(AD-04
- 部分失败处理:逐条 try/except,累计 error_detailsAD-10
- 新用户密码:secrets.token_hex(32) 生成 64 位随机哈希(AD-07
**覆盖 AC**: AC-13, AC-14, AC-16, AC-17, AC-18, AC-19, AC-20, AC-21, AC-22, AC-23, AC-24, AC-25, AC-26, AC-27, AC-28, AC-30
**依赖**: T-01, T-04, T-05, T-09, T-10
**验证**: mock Provider + mock DAO 的集成测试——完整同步流程 → 验证 DB 状态 + ChangeHandler 调用
---
### Phase 4: Infrastructure + API
#### T-12: Celery 任务 + Beat 定时调度
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/worker/org_sync/__init__.py`
- 新建: `src/backend/bisheng/worker/org_sync/tasks.py`
- 修改: `src/backend/bisheng/core/config/settings.py`
**内容**:
- execute_org_sync(config_id, trigger_type, trigger_user) Celery tasktime_limit=1800, soft_time_limit=1500
- check_org_sync_schedules() Beat task:每 60s 检查活跃 cron 配置,匹配时间则 dispatch execute_org_sync
- CeleryConf.task_routes 增加 `"bisheng.worker.org_sync.*": {"queue": "knowledge_celery"}`
- beat_schedule 增加 check_org_sync_schedulesschedule=60.0
- 使用 croniter 库解析 cron_expression 判断是否到期
- **tenant_id 传递**:发送任务时 tenant_id 通过 Celery headers 注入(`inject_tenant_header` signal),Worker 端 `before_task` signal 调用 `set_current_tenant_id()` 恢复到 ContextVarINV-8
**覆盖 AC**: AC-31, AC-32
**依赖**: T-11
**验证**: 单元测试——mock OrgSyncService,验证 task dispatch + cron 匹配逻辑
---
#### T-13: 请求/响应 DTO
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/domain/schemas/org_sync_schema.py`
**内容**:
- OrgSyncConfigCreateprovider, config_name, auth_type, auth_config(dict), sync_scope, schedule_type, cron_expression
- OrgSyncConfigUpdateauth_type, auth_config, sync_scope, schedule_type, cron_expression, status(全部 Optional
- OrgSyncConfigRead:完整字段,auth_config 脱敏后的 dict
- OrgSyncLogRead:完整字段
- RemoteTreeNodeexternal_id, name, children(递归)
- mask_sensitive_fields(auth_config: dict) → dict 脱敏函数
**覆盖 AC**: AC-34(脱敏逻辑)
**验证**: 单元测试——序列化/反序列化 + 脱敏函数验证
---
#### T-14: API 端点 — 配置 CRUD + 路由注册
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/__init__.py`
- 新建: `src/backend/bisheng/org_sync/api/__init__.py`
- 新建: `src/backend/bisheng/org_sync/api/router.py`
- 新建: `src/backend/bisheng/org_sync/api/endpoints/__init__.py`
- 新建: `src/backend/bisheng/org_sync/api/endpoints/sync_config.py`
- 修改: `src/backend/bisheng/api/router.py`
**内容**:
- 5 个配置 CRUD 端点(spec §6.1~6.5):
1. POST /org-sync/configs — 创建(加密 auth_config
2. GET /org-sync/configs — 列表
3. GET /org-sync/configs/{id} — 详情
4. PUT /org-sync/configs/{id} — 更新(合并 auth_config
5. DELETE /org-sync/configs/{id} — 软删除
- 模块 __init__.py 包文件(org_sync/ + api/ + endpoints/
- org_sync/api/router.py 路由聚合
- 权限检查:所有端点使用 `LoginUser.access_check` 要求管理员
- 响应包装:UnifiedResponseModelresp_200 / resp_500
- 路由注册到全局 `api/router.py`
**覆盖 AC**: AC-01, AC-02, AC-03, AC-04, AC-05, AC-06, AC-07, AC-08, AC-33, AC-34
**依赖**: T-01, T-03, T-13
**验证**: API 集成测试(TestClient)——配置 CRUD happy path + error path
---
#### T-15: API 端点 — 执行/测试/历史/远程树
- [x] 完成
**文件**:
- 新建: `src/backend/bisheng/org_sync/api/endpoints/sync_exec.py`
**内容**:
- 4 个执行端点(spec §6.6~6.9):
6. POST /org-sync/configs/{id}/test — 测试连接
7. POST /org-sync/configs/{id}/execute — 手动触发(dispatch Celery task
8. GET /org-sync/configs/{id}/logs — 同步历史(PageData
9. GET /org-sync/configs/{id}/remote-tree — 远程树预览
- 权限检查:同 T-14
- execute 端点:检查 sync_status + dispatch Celery task + 返回 log_id
- 路由注册到 org_sync/api/router.pyT-14 已创建)
**覆盖 AC**: AC-09, AC-10, AC-11, AC-12, AC-13, AC-14, AC-15, AC-29
**依赖**: T-01, T-03, T-05, T-11, T-12, T-14
**验证**: API 集成测试(TestClient)——测试连接 + 手动触发 + 历史查询 + 远程树
---
### Phase 5: Testing
#### T-16: Reconciler 单元测试
- [x] 完成
**文件**:
- 新建: `src/backend/test/test_org_sync_reconciler.py`
**内容**:
- test_dept_create:远程有新部门 → CreateDept
- test_dept_rename_third_party:第三方来源部门改名 → UpdateDept
- test_dept_rename_local:本地来源部门匹配远程 → UpdateDept + source 变更
- test_dept_move:部门层级变更 → MoveDept
- test_dept_archive:远程消失 → ArchiveDept
- test_dept_archive_cascade:归档含本地子部门 → 子部门也归档
- test_dept_topological_order:创建操作父先于子
- test_dept_cycle_detection:循环引用 → 跳过
- test_member_create:新员工 → CreateMember
- test_member_update:信息变更 → UpdateMember
- test_member_transfer:主部门变更 → TransferMember
- test_member_secondary_dept_change:附属部门增减
- test_member_disable:离职 → DisableMember
- test_member_reactivate:重新出现 → ReactivateMember
- test_member_local_conflict:本地用户匹配远程 → 强制覆盖
**覆盖 AC**: AC-16, AC-17, AC-18, AC-19, AC-20, AC-21, AC-22, AC-24, AC-25, AC-26, AC-27, AC-28
**依赖**: T-09, T-10
---
#### T-17: API 集成测试 + E2E 测试
- [x] 完成
**文件**:
- 新建: `src/backend/test/test_org_sync_api.py`
- 新建: `src/backend/test/e2e/test_e2e_org_sync.py`
**内容**:
- API 集成测试:
- test_create_config / test_create_duplicate / test_list_configs / test_get_config
- test_update_config_merge_auth / test_delete_config / test_cross_tenant_rejected
- test_test_connection_success / test_test_connection_auth_fail
- test_execute_sync / test_execute_already_running / test_execute_disabled
- test_get_logs_paginated / test_get_remote_tree
- test_permission_denied_non_admin
- E2E 测试(mock Provider,真实 DB):
- test_full_sync_flow:配置 → 触发 → 验证 Department/User/UserDepartment/OrgSyncLog 状态
- test_incremental_sync:首次同步后修改远程数据 → 二次同步 → 验证增量变更
- test_member_disable_on_departure:员工离职 → 验证禁用 + 清理
- test_multi_tenant_isolation:两个租户各自同步互不影响
**覆盖 AC**: AC-01, AC-02, AC-03, AC-04, AC-05, AC-06, AC-07, AC-08, AC-09, AC-10, AC-11, AC-12, AC-13, AC-14, AC-15, AC-16, AC-17, AC-18, AC-19, AC-20, AC-21, AC-22, AC-23, AC-24, AC-25, AC-26, AC-27, AC-28, AC-29, AC-30, AC-31, AC-32, AC-33, AC-34
**依赖**: T-14, T-15
---
## 任务依赖图
```
T-01 (ORM) ──┬──> T-02 (迁移)
├──> T-05 (Provider ABC) ──> T-06 (飞书)
│ ──> T-07 (通用API)
│ ──> T-08 (企微/钉钉 stub)
T-03 (错误码) ─┤
T-04 (User扩展)┤
├──> T-09 (部门Reconciler) ──┐
├──> T-10 (人员Reconciler) ──┤
│ v
└──> T-11 (OrgSyncService) <─┘
├──> T-12 (Celery)
T-13 (DTO)┤
v
T-14 (配置CRUD API)
v
T-15 (执行API) ───────> T-17 (API+E2E测试)
^
T-16 (Reconciler测试) ──────┘
```
## 并行策略
- **Wave 1** (可并行): T-01 + T-03 + T-04 + T-05
- **Wave 2** (T-01 完成后): T-02
- **Wave 3** (T-05 完成后,可并行): T-06 + T-07 + T-08
- **Wave 4** (T-01/T-04/T-09/T-10 完成后): T-11
- **Wave 5** (T-11 完成后,可并行): T-12 + T-13
- **Wave 6** (T-13 完成后): T-14
- **Wave 7** (T-14 完成后): T-15
- **Wave 8** (T-15 完成后,可并行): T-16 + T-17
-2
View File
@@ -150,8 +150,6 @@ F009-org-sync ← 仅依赖 F002P2,延后)
| **200** | **tenant (v2.5 新增)** | _待创建_ |
| **210** | **department (v2.5 新增)** | _待创建_ |
| **220** | **org_sync (v2.5 P2 新增)** | _待创建_ |
| **230** | **user_group (v2.5 新增)** | `common/errcode/user_group.py` |
| **240** | **role (v2.5 新增)** | `common/errcode/role.py` |
---
+2
View File
@@ -25,6 +25,7 @@ from bisheng.user_group.api.router import router as user_group_router
from bisheng.permission.api.router import router as permission_router
from bisheng.role.api.router import router as role_router
from bisheng.share_link.api.router import router as share_link_router
from bisheng.org_sync.api.router import router as org_sync_router
router = APIRouter(prefix='/api/v1', )
router.include_router(chat_router)
@@ -60,6 +61,7 @@ router.include_router(department_router)
router.include_router(user_group_router)
router.include_router(permission_router)
router.include_router(role_router)
router.include_router(org_sync_router)
router_rpc = APIRouter(prefix='/api/v2', )
router_rpc.include_router(knowledge_router_rpc)
@@ -0,0 +1,52 @@
from .base import BaseErrorCode
# Org sync module error codes, module code: 220
class OrgSyncConfigNotFoundError(BaseErrorCode):
Code: int = 22000
Msg: str = 'Org sync config not found'
class OrgSyncConfigDuplicateError(BaseErrorCode):
Code: int = 22001
Msg: str = 'Org sync config with same provider and name already exists'
class OrgSyncAuthFailedError(BaseErrorCode):
Code: int = 22002
Msg: str = 'Provider authentication failed'
class OrgSyncAlreadyRunningError(BaseErrorCode):
Code: int = 22003
Msg: str = 'Sync is already running for this config'
class OrgSyncProviderError(BaseErrorCode):
Code: int = 22004
Msg: str = 'Provider error'
class OrgSyncPermissionDeniedError(BaseErrorCode):
Code: int = 22005
Msg: str = 'No permission for org sync operation'
class OrgSyncInvalidConfigError(BaseErrorCode):
Code: int = 22006
Msg: str = 'Invalid org sync config'
class OrgSyncFetchError(BaseErrorCode):
Code: int = 22007
Msg: str = 'Failed to fetch data from provider'
class OrgSyncReconcileError(BaseErrorCode):
Code: int = 22008
Msg: str = 'Unrecoverable error during reconciliation'
class OrgSyncConfigDisabledError(BaseErrorCode):
Code: int = 22009
Msg: str = 'Org sync config is disabled'
@@ -146,6 +146,7 @@ class CeleryConf(BaseModel):
self.task_routers = {
"bisheng.worker.knowledge.*": {"queue": "knowledge_celery"}, # Knowledge Base Related Tasks
"bisheng.worker.workflow.*": {"queue": "workflow_celery"}, # Workflow Execution Related Tasks
"bisheng.worker.org_sync.*": {"queue": "knowledge_celery"}, # Org Sync Tasks (low frequency, reuse knowledge queue)
}
if 'telemetry_mid_user_increment' not in self.beat_schedule:
self.beat_schedule['telemetry_mid_user_increment'] = {
@@ -177,6 +178,11 @@ class CeleryConf(BaseModel):
'task': 'bisheng.worker.permission.retry_failed_tuples.retry_failed_tuples',
'schedule': 30.0, # Every 30 seconds
}
if 'check_org_sync_schedules' not in self.beat_schedule:
self.beat_schedule['check_org_sync_schedules'] = {
'task': 'bisheng.worker.org_sync.tasks.check_org_sync_schedules',
'schedule': 60.0, # Every 60 seconds
}
# convert str to crontab
for key, task_info in self.beat_schedule.items():
@@ -0,0 +1,131 @@
"""F009: Create org_sync_config, org_sync_log tables and extend user table.
Revision ID: f009_org_sync
Revises: f005_role_menu_quota
Create Date: 2026-04-13
Changes:
- CREATE TABLE org_sync_config (sync configuration per provider)
- CREATE TABLE org_sync_log (sync execution history)
- ALTER TABLE user ADD COLUMN source VARCHAR(32) NOT NULL DEFAULT 'local'
- ALTER TABLE user ADD COLUMN external_id VARCHAR(128) NULL
- CREATE UNIQUE INDEX uk_user_source_external_id ON user(source, external_id)
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = 'f009_org_sync'
down_revision: Union[str, Sequence[str], None] = 'f005_role_menu_quota'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# -- org_sync_config --
op.create_table(
'org_sync_config',
sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
sa.Column('tenant_id', sa.Integer, nullable=False, server_default='1',
comment='Tenant ID'),
sa.Column('provider', sa.String(32), nullable=False,
comment='Provider: feishu/wecom/dingtalk/generic_api'),
sa.Column('config_name', sa.String(128), nullable=False,
comment='User-given label'),
sa.Column('auth_type', sa.String(16), nullable=False,
comment='Auth mode: api_key/password'),
sa.Column('auth_config', sa.Text, nullable=False,
comment='Fernet-encrypted JSON'),
sa.Column('sync_scope', sa.JSON, nullable=True,
comment='Sync scope JSON'),
sa.Column('schedule_type', sa.String(16), nullable=False,
server_default='manual', comment='manual/cron'),
sa.Column('cron_expression', sa.String(64), nullable=True,
comment='Cron expression'),
sa.Column('sync_status', sa.String(16), nullable=False,
server_default='idle', comment='Runtime mutex: idle/running'),
sa.Column('last_sync_at', sa.DateTime, nullable=True,
comment='Last sync time'),
sa.Column('last_sync_result', sa.String(16), nullable=True,
comment='success/partial/failed'),
sa.Column('status', sa.String(16), nullable=False,
server_default='active', comment='active/disabled/deleted'),
sa.Column('create_user', sa.Integer, nullable=True,
comment='Creator user ID'),
sa.Column('create_time', sa.DateTime, nullable=False,
server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('update_time', sa.DateTime, nullable=False,
server_default=sa.text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')),
)
op.create_index('idx_osc_tenant', 'org_sync_config', ['tenant_id'])
op.create_index('idx_osc_status', 'org_sync_config', ['status'])
op.create_unique_constraint(
'uk_tenant_provider_name', 'org_sync_config',
['tenant_id', 'provider', 'config_name'],
)
# -- org_sync_log --
op.create_table(
'org_sync_log',
sa.Column('id', sa.BigInteger, primary_key=True, autoincrement=True),
sa.Column('tenant_id', sa.Integer, nullable=False, server_default='1'),
sa.Column('config_id', sa.Integer, nullable=False,
comment='FK to org_sync_config.id'),
sa.Column('trigger_type', sa.String(16), nullable=False,
comment='manual/scheduled'),
sa.Column('trigger_user', sa.Integer, nullable=True,
comment='User who triggered'),
sa.Column('status', sa.String(16), nullable=False,
server_default='running',
comment='running/success/partial/failed'),
sa.Column('dept_created', sa.Integer, nullable=False, server_default='0'),
sa.Column('dept_updated', sa.Integer, nullable=False, server_default='0'),
sa.Column('dept_archived', sa.Integer, nullable=False, server_default='0'),
sa.Column('member_created', sa.Integer, nullable=False, server_default='0'),
sa.Column('member_updated', sa.Integer, nullable=False, server_default='0'),
sa.Column('member_disabled', sa.Integer, nullable=False, server_default='0'),
sa.Column('member_reactivated', sa.Integer, nullable=False, server_default='0'),
sa.Column('error_details', sa.JSON, nullable=True,
comment='Error list JSON'),
sa.Column('start_time', sa.DateTime, nullable=True),
sa.Column('end_time', sa.DateTime, nullable=True),
sa.Column('create_time', sa.DateTime, nullable=False,
server_default=sa.text('CURRENT_TIMESTAMP')),
)
op.create_index('idx_osl_tenant', 'org_sync_log', ['tenant_id'])
op.create_index('idx_osl_config', 'org_sync_log', ['config_id'])
# -- user table extension --
op.add_column('user', sa.Column(
'source', sa.String(32), nullable=False,
server_default='local',
comment='Source: local/feishu/wecom/dingtalk/generic_api',
))
op.add_column('user', sa.Column(
'external_id', sa.String(128), nullable=True,
comment='External employee ID for sync',
))
op.create_unique_constraint(
'uk_user_source_external_id', 'user',
['source', 'external_id'],
)
def downgrade() -> None:
# -- user table rollback --
op.drop_constraint('uk_user_source_external_id', 'user', type_='unique')
op.drop_column('user', 'external_id')
op.drop_column('user', 'source')
# -- org_sync_log --
op.drop_index('idx_osl_config', table_name='org_sync_log')
op.drop_index('idx_osl_tenant', table_name='org_sync_log')
op.drop_table('org_sync_log')
# -- org_sync_config --
op.drop_constraint('uk_tenant_provider_name', 'org_sync_config', type_='unique')
op.drop_index('idx_osc_status', table_name='org_sync_config')
op.drop_index('idx_osc_tenant', table_name='org_sync_config')
op.drop_table('org_sync_config')
@@ -0,0 +1,188 @@
"""Org sync configuration CRUD endpoints (5).
Part of F009-org-sync. Spec §6.16.5.
"""
from fastapi import APIRouter, Depends
from bisheng.common.dependencies.user_deps import UserPayload
from bisheng.common.errcode.base import BaseErrorCode
from bisheng.common.errcode.org_sync import (
OrgSyncConfigNotFoundError,
OrgSyncInvalidConfigError,
OrgSyncPermissionDeniedError,
)
from bisheng.common.schemas.api import resp_200
from bisheng.org_sync.domain.models.org_sync import (
OrgSyncConfig,
OrgSyncConfigDao,
decrypt_auth_config,
encrypt_auth_config,
)
from bisheng.org_sync.domain.schemas.org_sync_schema import (
OrgSyncConfigCreate,
OrgSyncConfigRead,
OrgSyncConfigUpdate,
mask_sensitive_fields,
)
router = APIRouter()
def _config_to_read(config: OrgSyncConfig) -> dict:
"""Convert ORM model to response dict with masked auth_config."""
try:
auth_dict = decrypt_auth_config(config.auth_config)
except Exception:
auth_dict = {}
return OrgSyncConfigRead(
id=config.id,
provider=config.provider,
config_name=config.config_name,
auth_type=config.auth_type,
auth_config=mask_sensitive_fields(auth_dict),
sync_scope=config.sync_scope,
schedule_type=config.schedule_type,
cron_expression=config.cron_expression,
sync_status=config.sync_status,
last_sync_at=config.last_sync_at,
last_sync_result=config.last_sync_result,
status=config.status,
create_user=config.create_user,
create_time=config.create_time,
update_time=config.update_time,
).model_dump(mode='json')
@router.post('/configs')
async def create_config(
data: OrgSyncConfigCreate,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Create a new org sync configuration (spec §6.1, AC-01, AC-02)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = OrgSyncConfig(
tenant_id=login_user.tenant_id,
provider=data.provider,
config_name=data.config_name,
auth_type=data.auth_type,
auth_config=encrypt_auth_config(data.auth_config),
sync_scope=data.sync_scope,
schedule_type=data.schedule_type,
cron_expression=data.cron_expression,
create_user=login_user.user_id,
)
config = await OrgSyncConfigDao.acreate(config)
return resp_200(_config_to_read(config))
except BaseErrorCode as e:
return e.return_resp_instance()
except Exception as e:
if 'Duplicate entry' in str(e) or 'uk_tenant_provider_name' in str(e):
from bisheng.common.errcode.org_sync import OrgSyncConfigDuplicateError
return OrgSyncConfigDuplicateError.return_resp()
return OrgSyncInvalidConfigError.return_resp(msg=str(e))
@router.get('/configs')
async def list_configs(
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""List org sync configs for current tenant (spec §6.2, AC-03)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
configs = await OrgSyncConfigDao.aget_list(login_user.tenant_id)
return resp_200([_config_to_read(c) for c in configs])
except BaseErrorCode as e:
return e.return_resp_instance()
@router.get('/configs/{config_id}')
async def get_config(
config_id: int,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Get org sync config details (spec §6.3, AC-04, AC-08)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
if config.status == 'deleted':
return OrgSyncConfigNotFoundError.return_resp()
return resp_200(_config_to_read(config))
except BaseErrorCode as e:
return e.return_resp_instance()
@router.put('/configs/{config_id}')
async def update_config(
config_id: int,
data: OrgSyncConfigUpdate,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Update org sync config (spec §6.4, AC-05, AC-06)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
if config.status == 'deleted':
return OrgSyncConfigNotFoundError.return_resp()
# Merge auth_config (AC-06)
if data.auth_config is not None:
existing_auth = decrypt_auth_config(config.auth_config)
existing_auth.update(data.auth_config)
config.auth_config = encrypt_auth_config(existing_auth)
if data.auth_type is not None:
config.auth_type = data.auth_type
if data.sync_scope is not None:
config.sync_scope = data.sync_scope
if data.schedule_type is not None:
config.schedule_type = data.schedule_type
if data.cron_expression is not None:
config.cron_expression = data.cron_expression
if data.status is not None:
config.status = data.status
if data.config_name is not None:
config.config_name = data.config_name
config = await OrgSyncConfigDao.aupdate(config)
return resp_200(_config_to_read(config))
except BaseErrorCode as e:
return e.return_resp_instance()
@router.delete('/configs/{config_id}')
async def delete_config(
config_id: int,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Soft-delete org sync config (spec §6.5, AC-07)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
if config.sync_status == 'running':
from bisheng.common.errcode.org_sync import OrgSyncAlreadyRunningError
return OrgSyncAlreadyRunningError.return_resp()
config.status = 'deleted'
await OrgSyncConfigDao.aupdate(config)
return resp_200(None)
except BaseErrorCode as e:
return e.return_resp_instance()
@@ -0,0 +1,188 @@
"""Org sync execution endpoints (4): test, execute, logs, remote-tree.
Part of F009-org-sync. Spec §6.66.9.
"""
import asyncio
from fastapi import APIRouter, Depends, Query
from bisheng.common.dependencies.user_deps import UserPayload
from bisheng.common.errcode.base import BaseErrorCode
from bisheng.common.errcode.org_sync import (
OrgSyncAlreadyRunningError,
OrgSyncConfigDisabledError,
OrgSyncConfigNotFoundError,
OrgSyncPermissionDeniedError,
)
from bisheng.common.schemas.api import PageData, resp_200
from bisheng.org_sync.domain.models.org_sync import (
OrgSyncConfigDao,
OrgSyncLog,
OrgSyncLogDao,
decrypt_auth_config,
)
from bisheng.org_sync.domain.providers.base import get_provider
from bisheng.org_sync.domain.schemas.org_sync_schema import OrgSyncLogRead, RemoteTreeNode
router = APIRouter()
@router.post('/configs/{config_id}/test')
async def test_connection(
config_id: int,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Test provider connectivity (spec §6.6, AC-09, AC-10)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
auth_config = decrypt_auth_config(config.auth_config)
provider = get_provider(config.provider, auth_config)
result = await provider.test_connection()
return resp_200(result)
except BaseErrorCode as e:
return e.return_resp_instance()
@router.post('/configs/{config_id}/execute')
async def execute_sync(
config_id: int,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Manually trigger sync execution (spec §6.7, AC-13, AC-14, AC-15)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
if config.status == 'disabled':
return OrgSyncConfigDisabledError.return_resp()
if config.sync_status == 'running':
return OrgSyncAlreadyRunningError.return_resp()
# Create log entry first, then dispatch Celery task
from datetime import datetime
log = OrgSyncLog(
tenant_id=config.tenant_id,
config_id=config_id,
trigger_type='manual',
trigger_user=login_user.user_id,
status='running',
start_time=datetime.now(),
)
log = await OrgSyncLogDao.acreate(log)
from bisheng.worker.org_sync.tasks import execute_org_sync
execute_org_sync.apply_async(
args=[config_id, 'manual', login_user.user_id],
queue='knowledge_celery',
)
return resp_200({
'log_id': log.id,
'message': 'Sync task dispatched',
})
except BaseErrorCode as e:
return e.return_resp_instance()
@router.get('/configs/{config_id}/logs')
async def get_sync_logs(
config_id: int,
page: int = Query(default=1, ge=1),
limit: int = Query(default=20, ge=1, le=100),
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Get sync history logs with pagination (spec §6.8, AC-29, AC-30)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
logs, total = await OrgSyncLogDao.aget_by_config(config_id, page, limit)
log_reads = [
OrgSyncLogRead(
id=log.id,
config_id=log.config_id,
trigger_type=log.trigger_type,
trigger_user=log.trigger_user,
status=log.status,
dept_created=log.dept_created,
dept_updated=log.dept_updated,
dept_archived=log.dept_archived,
member_created=log.member_created,
member_updated=log.member_updated,
member_disabled=log.member_disabled,
member_reactivated=log.member_reactivated,
error_details=log.error_details,
start_time=log.start_time,
end_time=log.end_time,
create_time=log.create_time,
).model_dump(mode='json')
for log in logs
]
return resp_200(PageData(data=log_reads, total=total).model_dump())
except BaseErrorCode as e:
return e.return_resp_instance()
@router.get('/configs/{config_id}/remote-tree')
async def get_remote_tree(
config_id: int,
login_user: UserPayload = Depends(UserPayload.get_login_user),
):
"""Preview remote org tree from provider (spec §6.9, AC-11, AC-12)."""
if not login_user.is_admin():
return OrgSyncPermissionDeniedError.return_resp()
try:
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config or config.tenant_id != login_user.tenant_id:
return OrgSyncConfigNotFoundError.return_resp()
auth_config = decrypt_auth_config(config.auth_config)
provider = get_provider(config.provider, auth_config)
await provider.authenticate()
scope = config.sync_scope
root_dept_ids = scope.get('root_dept_ids') if scope else None
remote_depts = await provider.fetch_departments(root_dept_ids)
# Build tree structure
tree = _build_tree(remote_depts)
return resp_200([node.model_dump() for node in tree])
except BaseErrorCode as e:
return e.return_resp_instance()
def _build_tree(depts) -> list[RemoteTreeNode]:
"""Convert flat department list to nested tree."""
nodes: dict[str, RemoteTreeNode] = {}
for d in depts:
nodes[d.external_id] = RemoteTreeNode(
external_id=d.external_id,
name=d.name,
)
roots: list[RemoteTreeNode] = []
for d in depts:
node = nodes[d.external_id]
if d.parent_external_id and d.parent_external_id in nodes:
nodes[d.parent_external_id].children.append(node)
else:
roots.append(node)
return roots
@@ -0,0 +1,13 @@
"""Org sync module router aggregation.
Part of F009-org-sync.
"""
from fastapi import APIRouter
from bisheng.org_sync.api.endpoints.sync_config import router as config_router
from bisheng.org_sync.api.endpoints.sync_exec import router as exec_router
router = APIRouter(prefix='/org-sync', tags=['Org Sync'])
router.include_router(config_router)
router.include_router(exec_router)
@@ -0,0 +1,373 @@
"""OrgSyncConfig and OrgSyncLog ORM models + DAO classes.
Part of F009-org-sync.
"""
import json
from datetime import datetime
from typing import List, Optional, Tuple
from sqlalchemy import (
BigInteger,
Column,
DateTime,
Integer,
JSON,
String,
Text,
UniqueConstraint,
func,
text,
update,
)
from sqlmodel import Field, select
from bisheng.common.models.base import SQLModelSerializable
from bisheng.core.config.settings import decrypt_token, encrypt_token
from bisheng.core.database import get_async_db_session
# ---------------------------------------------------------------------------
# Encryption helpers for auth_config (Fernet, AD-02)
# ---------------------------------------------------------------------------
def encrypt_auth_config(config_dict: dict) -> str:
"""Encrypt auth_config dict to a Fernet-encrypted string for DB storage."""
raw = json.dumps(config_dict, ensure_ascii=False)
encrypted_bytes = encrypt_token(raw)
# encrypt_token returns bytes; decode to str for TEXT column
return encrypted_bytes.decode() if isinstance(encrypted_bytes, bytes) else encrypted_bytes
def decrypt_auth_config(encrypted: str) -> dict:
"""Decrypt a Fernet-encrypted auth_config string back to a dict."""
raw = decrypt_token(encrypted)
return json.loads(raw)
# ---------------------------------------------------------------------------
# ORM Models
# ---------------------------------------------------------------------------
class OrgSyncConfig(SQLModelSerializable, table=True):
__tablename__ = 'org_sync_config'
id: Optional[int] = Field(
default=None,
sa_column=Column(Integer, primary_key=True, autoincrement=True),
)
tenant_id: int = Field(
default=1,
sa_column=Column(
Integer, nullable=False,
server_default=text('1'), index=True,
comment='Tenant ID',
),
)
provider: str = Field(
sa_column=Column(
String(32), nullable=False,
comment='Provider: feishu/wecom/dingtalk/generic_api',
),
)
config_name: str = Field(
sa_column=Column(
String(128), nullable=False,
comment='User-given label, e.g. Feishu Production',
),
)
auth_type: str = Field(
sa_column=Column(
String(16), nullable=False,
comment='Auth mode: api_key/password (oauth reserved)',
),
)
auth_config: str = Field(
sa_column=Column(
Text, nullable=False,
comment='Fernet-encrypted JSON: credentials per auth_type',
),
)
sync_scope: Optional[dict] = Field(
default=None,
sa_column=Column(
JSON, nullable=True,
comment='Sync scope: {"root_dept_ids": ["id1","id2"]} or null=all',
),
)
schedule_type: str = Field(
default='manual',
sa_column=Column(
String(16), nullable=False,
server_default=text("'manual'"),
comment='Execution mode: manual/cron',
),
)
cron_expression: Optional[str] = Field(
default=None,
sa_column=Column(
String(64), nullable=True,
comment='Cron expression, e.g. 0 2 * * *',
),
)
sync_status: str = Field(
default='idle',
sa_column=Column(
String(16), nullable=False,
server_default=text("'idle'"),
comment='Runtime mutex: idle/running',
),
)
last_sync_at: Optional[datetime] = Field(
default=None,
sa_column=Column(DateTime, nullable=True, comment='Last sync time'),
)
last_sync_result: Optional[str] = Field(
default=None,
sa_column=Column(
String(16), nullable=True,
comment='Last sync result: success/partial/failed',
),
)
status: str = Field(
default='active',
sa_column=Column(
String(16), nullable=False,
server_default=text("'active'"), index=True,
comment='Config status: active/disabled/deleted',
),
)
create_user: Optional[int] = Field(
default=None,
sa_column=Column(Integer, nullable=True, comment='Creator user ID'),
)
create_time: Optional[datetime] = Field(
default=None,
sa_column=Column(
DateTime, nullable=False,
server_default=text('CURRENT_TIMESTAMP'),
),
)
update_time: Optional[datetime] = Field(
default=None,
sa_column=Column(
DateTime, nullable=False,
server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'),
),
)
__table_args__ = (
UniqueConstraint(
'tenant_id', 'provider', 'config_name',
name='uk_tenant_provider_name',
),
)
class OrgSyncLog(SQLModelSerializable, table=True):
__tablename__ = 'org_sync_log'
id: Optional[int] = Field(
default=None,
sa_column=Column(BigInteger, primary_key=True, autoincrement=True),
)
tenant_id: int = Field(
default=1,
sa_column=Column(
Integer, nullable=False,
server_default=text('1'), index=True,
),
)
config_id: int = Field(
sa_column=Column(
Integer, nullable=False, index=True,
comment='FK to org_sync_config.id',
),
)
trigger_type: str = Field(
sa_column=Column(
String(16), nullable=False,
comment='Trigger: manual/scheduled',
),
)
trigger_user: Optional[int] = Field(
default=None,
sa_column=Column(
Integer, nullable=True,
comment='User who triggered (null for scheduled)',
),
)
status: str = Field(
default='running',
sa_column=Column(
String(16), nullable=False,
server_default=text("'running'"),
comment='Status: running/success/partial/failed',
),
)
dept_created: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
dept_updated: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
dept_archived: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
member_created: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
member_updated: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
member_disabled: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
member_reactivated: int = Field(
default=0,
sa_column=Column(Integer, nullable=False, server_default=text('0')),
)
error_details: Optional[list] = Field(
default=None,
sa_column=Column(
JSON, nullable=True,
comment='Error list: [{entity_type, external_id, error_msg}]',
),
)
start_time: Optional[datetime] = Field(
default=None,
sa_column=Column(DateTime, nullable=True),
)
end_time: Optional[datetime] = Field(
default=None,
sa_column=Column(DateTime, nullable=True),
)
create_time: Optional[datetime] = Field(
default=None,
sa_column=Column(
DateTime, nullable=False,
server_default=text('CURRENT_TIMESTAMP'),
),
)
# ---------------------------------------------------------------------------
# DAO Classes
# ---------------------------------------------------------------------------
class OrgSyncConfigDao:
@classmethod
async def acreate(cls, config: OrgSyncConfig) -> OrgSyncConfig:
async with get_async_db_session() as session:
session.add(config)
await session.flush()
await session.refresh(config)
await session.commit()
return config
@classmethod
async def aget_by_id(cls, config_id: int) -> Optional[OrgSyncConfig]:
async with get_async_db_session() as session:
statement = select(OrgSyncConfig).where(
OrgSyncConfig.id == config_id,
)
result = await session.exec(statement)
return result.first()
@classmethod
async def aget_list(
cls, tenant_id: int, status: str = 'active',
) -> List[OrgSyncConfig]:
async with get_async_db_session() as session:
statement = select(OrgSyncConfig).where(
OrgSyncConfig.tenant_id == tenant_id,
OrgSyncConfig.status != 'deleted',
)
if status:
statement = statement.where(OrgSyncConfig.status == status)
statement = statement.order_by(OrgSyncConfig.id.desc())
result = await session.exec(statement)
return result.all()
@classmethod
async def aupdate(cls, config: OrgSyncConfig) -> OrgSyncConfig:
async with get_async_db_session() as session:
session.add(config)
await session.commit()
await session.refresh(config)
return config
@classmethod
async def aset_sync_status(
cls, config_id: int, old_status: str, new_status: str,
) -> bool:
"""Atomic CAS update of sync_status. Returns True if row was updated."""
async with get_async_db_session() as session:
stmt = (
update(OrgSyncConfig)
.where(
OrgSyncConfig.id == config_id,
OrgSyncConfig.sync_status == old_status,
)
.values(sync_status=new_status)
)
result = await session.execute(stmt)
await session.commit()
return result.rowcount > 0
@classmethod
async def aget_active_cron_configs(cls) -> List[OrgSyncConfig]:
"""Get all configs with schedule_type='cron' and status='active'."""
async with get_async_db_session() as session:
statement = select(OrgSyncConfig).where(
OrgSyncConfig.schedule_type == 'cron',
OrgSyncConfig.status == 'active',
)
result = await session.exec(statement)
return result.all()
class OrgSyncLogDao:
@classmethod
async def acreate(cls, log: OrgSyncLog) -> OrgSyncLog:
async with get_async_db_session() as session:
session.add(log)
await session.flush()
await session.refresh(log)
await session.commit()
return log
@classmethod
async def aupdate(cls, log: OrgSyncLog) -> OrgSyncLog:
async with get_async_db_session() as session:
session.add(log)
await session.commit()
await session.refresh(log)
return log
@classmethod
async def aget_by_config(
cls, config_id: int, page: int = 1, limit: int = 20,
) -> Tuple[List[OrgSyncLog], int]:
"""Paginated query of logs for a given config, newest first."""
async with get_async_db_session() as session:
base = select(OrgSyncLog).where(
OrgSyncLog.config_id == config_id,
)
# Total count
count_stmt = select(func.count()).select_from(base.subquery())
total = await session.scalar(count_stmt) or 0
# Page data
data_stmt = base.order_by(OrgSyncLog.id.desc())
if page and limit:
data_stmt = data_stmt.offset((page - 1) * limit).limit(limit)
result = await session.exec(data_stmt)
return result.all(), total
@@ -0,0 +1,83 @@
"""OrgSyncProvider abstract base class and factory."""
import importlib
from abc import ABC, abstractmethod
from typing import Optional
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
class OrgSyncProvider(ABC):
"""Abstract base for third-party org data providers.
Subclasses implement four methods to integrate a specific platform
(Feishu, WeCom, DingTalk, or a generic REST API).
"""
def __init__(self, auth_config: dict):
self.auth_config = auth_config
@abstractmethod
async def authenticate(self) -> bool:
"""Validate credentials and obtain access token if needed.
Returns True on success; raises OrgSyncAuthFailedError on failure.
"""
...
@abstractmethod
async def fetch_departments(
self, root_dept_ids: Optional[list[str]] = None,
) -> list[RemoteDepartmentDTO]:
"""Fetch department tree from the provider.
Args:
root_dept_ids: Optional scope filter. None = fetch all.
"""
...
@abstractmethod
async def fetch_members(
self, department_ids: Optional[list[str]] = None,
) -> list[RemoteMemberDTO]:
"""Fetch members from the provider.
Args:
department_ids: Departments to fetch members from. None = all.
"""
...
@abstractmethod
async def test_connection(self) -> dict:
"""Test connectivity and return summary info.
Returns dict with keys: connected (bool), org_name (str),
total_depts (int), total_members (int).
"""
...
# Provider registry: lazy imports to avoid loading unused providers
_PROVIDER_REGISTRY = {
'feishu': 'bisheng.org_sync.domain.providers.feishu.FeishuProvider',
'wecom': 'bisheng.org_sync.domain.providers.wecom.WeComProvider',
'dingtalk': 'bisheng.org_sync.domain.providers.dingtalk.DingTalkProvider',
'generic_api': 'bisheng.org_sync.domain.providers.generic_api.GenericAPIProvider',
}
def get_provider(provider: str, auth_config: dict) -> OrgSyncProvider:
"""Factory: instantiate the correct provider by name.
Raises OrgSyncProviderError for unknown/unimplemented providers.
"""
from bisheng.common.errcode.org_sync import OrgSyncProviderError
dotted_path = _PROVIDER_REGISTRY.get(provider)
if not dotted_path:
raise OrgSyncProviderError(msg=f'Unknown provider: {provider}')
module_path, class_name = dotted_path.rsplit('.', 1)
module = importlib.import_module(module_path)
cls = getattr(module, class_name)
return cls(auth_config)
@@ -0,0 +1,35 @@
"""DingTalk Provider — stub implementation.
API Reference: https://open.dingtalk.com/document/orgapp/obtain-the-department-list-v2
Authentication: app_key + app_secret → access_token
Departments: POST /topapi/v2/department/listsub
Members: POST /topapi/v2/user/list
Full implementation deferred to a future release.
"""
from typing import Optional
from bisheng.common.errcode.org_sync import OrgSyncProviderError
from bisheng.org_sync.domain.providers.base import OrgSyncProvider
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
class DingTalkProvider(OrgSyncProvider):
"""DingTalk provider — not yet implemented."""
async def authenticate(self) -> bool:
raise OrgSyncProviderError(msg='DingTalk provider not implemented')
async def fetch_departments(
self, root_dept_ids: Optional[list[str]] = None,
) -> list[RemoteDepartmentDTO]:
raise OrgSyncProviderError(msg='DingTalk provider not implemented')
async def fetch_members(
self, department_ids: Optional[list[str]] = None,
) -> list[RemoteMemberDTO]:
raise OrgSyncProviderError(msg='DingTalk provider not implemented')
async def test_connection(self) -> dict:
raise OrgSyncProviderError(msg='DingTalk provider not implemented')
@@ -0,0 +1,260 @@
"""Feishu (Lark) Provider — full implementation.
Uses Feishu Open Platform Contact API v3:
- Auth: POST /auth/v3/tenant_access_token/internal
- Departments: GET /contact/v3/departments/{dept_id}/children (BFS)
- Members: GET /contact/v3/users?department_id=X (page_token pagination)
Rate limit: asyncio.Semaphore(5) + 429 exponential backoff (1s/2s/4s).
Token cache: 2-hour TTL (Feishu token validity).
"""
import asyncio
import time
from typing import Optional
import httpx
from loguru import logger
from bisheng.common.errcode.org_sync import OrgSyncAuthFailedError, OrgSyncFetchError
from bisheng.org_sync.domain.providers.base import OrgSyncProvider
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
FEISHU_BASE_URL = 'https://open.feishu.cn/open-apis'
TOKEN_TTL_SECONDS = 7200 # 2 hours
MAX_RETRIES = 3
BACKOFF_BASE = 1 # seconds
class FeishuProvider(OrgSyncProvider):
def __init__(self, auth_config: dict):
super().__init__(auth_config)
self._token: Optional[str] = None
self._token_expires_at: float = 0
self._semaphore = asyncio.Semaphore(5)
# ------------------------------------------------------------------
# HTTP helper with rate-limit retry
# ------------------------------------------------------------------
async def _request(
self,
client: httpx.AsyncClient,
method: str,
url: str,
**kwargs,
) -> dict:
"""Send an HTTP request with semaphore throttling and 429 backoff."""
async with self._semaphore:
for attempt in range(MAX_RETRIES + 1):
resp = await client.request(method, url, **kwargs)
if resp.status_code == 429:
if attempt < MAX_RETRIES:
wait = BACKOFF_BASE * (2 ** attempt)
logger.warning(f'Feishu 429 rate-limited, retrying in {wait}s')
await asyncio.sleep(wait)
continue
raise OrgSyncFetchError(msg='Feishu API rate limit exceeded after retries')
resp.raise_for_status()
data = resp.json()
if data.get('code', 0) != 0:
raise OrgSyncFetchError(
msg=f"Feishu API error: {data.get('msg', 'unknown')} (code={data.get('code')})",
)
return data
# Should not reach here
raise OrgSyncFetchError(msg='Feishu request failed')
# ------------------------------------------------------------------
# Token management
# ------------------------------------------------------------------
async def _ensure_token(self, client: httpx.AsyncClient) -> str:
"""Return a valid tenant_access_token, refreshing if expired."""
if self._token and time.time() < self._token_expires_at:
return self._token
app_id = self.auth_config.get('app_id', '')
app_secret = self.auth_config.get('app_secret', '')
if not app_id or not app_secret:
raise OrgSyncAuthFailedError(msg='Missing app_id or app_secret in auth_config')
resp = await client.post(
f'{FEISHU_BASE_URL}/auth/v3/tenant_access_token/internal',
json={'app_id': app_id, 'app_secret': app_secret},
)
resp.raise_for_status()
body = resp.json()
if body.get('code', -1) != 0:
raise OrgSyncAuthFailedError(
msg=f"Feishu auth failed: {body.get('msg', 'unknown')}",
)
self._token = body['tenant_access_token']
expire = body.get('expire', TOKEN_TTL_SECONDS)
self._token_expires_at = time.time() + expire - 60 # refresh 1 min early
return self._token
def _auth_headers(self, token: str) -> dict:
return {'Authorization': f'Bearer {token}'}
# ------------------------------------------------------------------
# Public interface
# ------------------------------------------------------------------
async def authenticate(self) -> bool:
async with httpx.AsyncClient(timeout=30) as client:
await self._ensure_token(client)
return True
async def fetch_departments(
self, root_dept_ids: Optional[list[str]] = None,
) -> list[RemoteDepartmentDTO]:
"""BFS traversal of the Feishu department tree."""
results: list[RemoteDepartmentDTO] = []
async with httpx.AsyncClient(timeout=30) as client:
token = await self._ensure_token(client)
headers = self._auth_headers(token)
# Starting points: provided root IDs or Feishu root "0"
queue = list(root_dept_ids) if root_dept_ids else ['0']
visited: set[str] = set()
while queue:
parent_id = queue.pop(0)
if parent_id in visited:
continue
visited.add(parent_id)
page_token: Optional[str] = None
while True:
params = {
'department_id_type': 'open_department_id',
'parent_department_id': parent_id,
'page_size': 50,
}
if page_token:
params['page_token'] = page_token
data = await self._request(
client, 'GET',
f'{FEISHU_BASE_URL}/contact/v3/departments/{parent_id}/children',
headers=headers,
params=params,
)
items = data.get('data', {}).get('items', [])
for item in items:
dept_id = item.get('open_department_id', '')
results.append(RemoteDepartmentDTO(
external_id=dept_id,
name=item.get('name', ''),
parent_external_id=parent_id if parent_id != '0' else None,
sort_order=int(item.get('order', '0') or '0'),
))
queue.append(dept_id)
has_more = data.get('data', {}).get('has_more', False)
page_token = data.get('data', {}).get('page_token')
if not has_more or not page_token:
break
return results
async def fetch_members(
self, department_ids: Optional[list[str]] = None,
) -> list[RemoteMemberDTO]:
"""Fetch members from specified departments (or all)."""
results: list[RemoteMemberDTO] = []
seen_user_ids: set[str] = set()
async with httpx.AsyncClient(timeout=30) as client:
token = await self._ensure_token(client)
headers = self._auth_headers(token)
if not department_ids:
department_ids = ['0']
for dept_id in department_ids:
page_token: Optional[str] = None
while True:
params = {
'department_id_type': 'open_department_id',
'department_id': dept_id,
'page_size': 50,
}
if page_token:
params['page_token'] = page_token
data = await self._request(
client, 'GET',
f'{FEISHU_BASE_URL}/contact/v3/users',
headers=headers,
params=params,
)
items = data.get('data', {}).get('items', [])
for item in items:
user_id = item.get('open_id', '') or item.get('user_id', '')
if user_id in seen_user_ids:
continue
seen_user_ids.add(user_id)
dept_ids_list = item.get('department_ids', [])
primary_dept = dept_ids_list[0] if dept_ids_list else dept_id
secondary_depts = [d for d in dept_ids_list[1:] if d != primary_dept]
status = 'active'
if item.get('status', {}).get('is_frozen', False):
status = 'disabled'
if item.get('status', {}).get('is_resigned', False):
status = 'disabled'
results.append(RemoteMemberDTO(
external_id=user_id,
name=item.get('name', ''),
email=item.get('email'),
phone=item.get('mobile'),
primary_dept_external_id=primary_dept,
secondary_dept_external_ids=secondary_depts,
status=status,
))
has_more = data.get('data', {}).get('has_more', False)
page_token = data.get('data', {}).get('page_token')
if not has_more or not page_token:
break
return results
async def test_connection(self) -> dict:
"""Test connectivity: authenticate + fetch root department info."""
async with httpx.AsyncClient(timeout=30) as client:
token = await self._ensure_token(client)
headers = self._auth_headers(token)
data = await self._request(
client, 'GET',
f'{FEISHU_BASE_URL}/contact/v3/departments/0',
headers=headers,
params={'department_id_type': 'open_department_id'},
)
dept_info = data.get('data', {}).get('department', {})
# Get a count of child departments for summary
children_data = await self._request(
client, 'GET',
f'{FEISHU_BASE_URL}/contact/v3/departments/0/children',
headers=headers,
params={
'department_id_type': 'open_department_id',
'parent_department_id': '0',
'page_size': 1,
},
)
return {
'connected': True,
'org_name': dept_info.get('name', 'Unknown'),
'total_depts': children_data.get('data', {}).get('total', 0),
'total_members': dept_info.get('member_count', 0),
}
@@ -0,0 +1,206 @@
"""Generic REST API Provider — full implementation.
Supports any third-party system that exposes department/member data via REST.
Configuration is entirely in auth_config:
- departments_url / members_url: endpoint URLs
- api_key + param_location: for header/query auth
- server_addr + username + password: for basic auth
- field_mapping: maps provider-specific field names to standard DTO fields
"""
from typing import Optional
import httpx
from loguru import logger
from bisheng.common.errcode.org_sync import OrgSyncAuthFailedError, OrgSyncFetchError
from bisheng.org_sync.domain.providers.base import OrgSyncProvider
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
# Default field mapping if not provided
DEFAULT_FIELD_MAPPING = {
'dept_id': 'id',
'dept_name': 'name',
'dept_parent_id': 'parentId',
'member_id': 'employeeId',
'member_name': 'fullName',
'member_email': 'email',
'member_phone': 'mobile',
'member_primary_dept': 'mainDepartment',
'member_secondary_depts': 'otherDepartments',
'member_status': 'status',
}
class GenericAPIProvider(OrgSyncProvider):
def __init__(self, auth_config: dict):
super().__init__(auth_config)
self._field_mapping = {
**DEFAULT_FIELD_MAPPING,
**auth_config.get('field_mapping', {}),
}
# ------------------------------------------------------------------
# HTTP helper
# ------------------------------------------------------------------
def _build_auth(self) -> tuple[dict, httpx.BasicAuth | None]:
"""Build auth headers/params and optional BasicAuth from config."""
headers: dict = {}
params: dict = {}
basic_auth: httpx.BasicAuth | None = None
api_key = self.auth_config.get('api_key')
if api_key:
location = self.auth_config.get('param_location', 'header')
if location == 'query':
params['api_key'] = api_key
else:
headers['Authorization'] = f'Bearer {api_key}'
else:
username = self.auth_config.get('username', '')
password = self.auth_config.get('password', '')
if username:
basic_auth = httpx.BasicAuth(username, password)
return headers, basic_auth
async def _fetch_json(self, client: httpx.AsyncClient, url: str) -> dict:
"""Fetch JSON from a URL with configured auth."""
headers, basic_auth = self._build_auth()
try:
resp = await client.get(url, headers=headers, auth=basic_auth)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
raise OrgSyncFetchError(
msg=f'Generic API HTTP error {e.response.status_code}: {url}',
)
except Exception as e:
raise OrgSyncFetchError(msg=f'Generic API request failed: {e}')
def _get_field(self, item: dict, mapping_key: str, default=None):
"""Extract a field from item using the configured field mapping."""
field_name = self._field_mapping.get(mapping_key, '')
return item.get(field_name, default)
# ------------------------------------------------------------------
# Public interface
# ------------------------------------------------------------------
async def authenticate(self) -> bool:
"""Validate that we can reach at least one configured URL."""
test_url = (
self.auth_config.get('departments_url')
or self.auth_config.get('members_url')
or self.auth_config.get('server_addr')
or self.auth_config.get('endpoint_url')
)
if not test_url:
raise OrgSyncAuthFailedError(msg='No endpoint URL configured')
async with httpx.AsyncClient(timeout=30) as client:
headers, basic_auth = self._build_auth()
try:
resp = await client.get(test_url, headers=headers, auth=basic_auth)
resp.raise_for_status()
except Exception as e:
raise OrgSyncAuthFailedError(msg=f'Authentication failed: {e}')
return True
async def fetch_departments(
self, root_dept_ids: Optional[list[str]] = None,
) -> list[RemoteDepartmentDTO]:
url = self.auth_config.get('departments_url')
if not url:
raise OrgSyncFetchError(msg='departments_url not configured')
async with httpx.AsyncClient(timeout=60) as client:
data = await self._fetch_json(client, url)
# Accept {"departments": [...]} or top-level list
items = data if isinstance(data, list) else data.get('departments', [])
if not isinstance(items, list):
raise OrgSyncFetchError(
msg=f'Unexpected departments response format: {type(items).__name__}',
)
results = []
for item in items:
ext_id = str(self._get_field(item, 'dept_id', ''))
if not ext_id:
logger.warning(f'Skipping department with missing ID: {item}')
continue
parent_id = self._get_field(item, 'dept_parent_id')
if parent_id is not None:
parent_id = str(parent_id) if parent_id else None
# Apply root scope filter if provided
if root_dept_ids and ext_id not in root_dept_ids and parent_id not in root_dept_ids:
continue
results.append(RemoteDepartmentDTO(
external_id=ext_id,
name=str(self._get_field(item, 'dept_name', '')),
parent_external_id=parent_id,
sort_order=int(self._get_field(item, 'sort_order', 0) or 0),
))
return results
async def fetch_members(
self, department_ids: Optional[list[str]] = None,
) -> list[RemoteMemberDTO]:
url = self.auth_config.get('members_url')
if not url:
raise OrgSyncFetchError(msg='members_url not configured')
async with httpx.AsyncClient(timeout=60) as client:
data = await self._fetch_json(client, url)
# Accept {"members": [...]} or top-level list
items = data if isinstance(data, list) else data.get('members', [])
if not isinstance(items, list):
raise OrgSyncFetchError(
msg=f'Unexpected members response format: {type(items).__name__}',
)
results = []
for item in items:
ext_id = str(self._get_field(item, 'member_id', ''))
if not ext_id:
logger.warning(f'Skipping member with missing ID: {item}')
continue
primary_dept = str(self._get_field(item, 'member_primary_dept', '') or '')
secondary = self._get_field(item, 'member_secondary_depts', []) or []
if isinstance(secondary, str):
secondary = [s.strip() for s in secondary.split(',') if s.strip()]
secondary = [str(s) for s in secondary]
status_val = str(self._get_field(item, 'member_status', 'active') or 'active')
results.append(RemoteMemberDTO(
external_id=ext_id,
name=str(self._get_field(item, 'member_name', '')),
email=self._get_field(item, 'member_email'),
phone=self._get_field(item, 'member_phone'),
primary_dept_external_id=primary_dept,
secondary_dept_external_ids=secondary,
status=status_val,
))
return results
async def test_connection(self) -> dict:
async with httpx.AsyncClient(timeout=30) as client:
depts = await self.fetch_departments()
members = await self.fetch_members()
return {
'connected': True,
'org_name': 'Generic API',
'total_depts': len(depts),
'total_members': len(members),
}
@@ -0,0 +1,35 @@
"""WeChat Work (WeCom) Provider — stub implementation.
API Reference: https://developer.work.weixin.qq.com/document/path/90208
Authentication: corp_id + corp_secret → access_token
Departments: GET /cgi-bin/department/list
Members: GET /cgi-bin/user/list?department_id=X
Full implementation deferred to a future release.
"""
from typing import Optional
from bisheng.common.errcode.org_sync import OrgSyncProviderError
from bisheng.org_sync.domain.providers.base import OrgSyncProvider
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
class WeComProvider(OrgSyncProvider):
"""WeChat Work provider — not yet implemented."""
async def authenticate(self) -> bool:
raise OrgSyncProviderError(msg='WeChat Work provider not implemented')
async def fetch_departments(
self, root_dept_ids: Optional[list[str]] = None,
) -> list[RemoteDepartmentDTO]:
raise OrgSyncProviderError(msg='WeChat Work provider not implemented')
async def fetch_members(
self, department_ids: Optional[list[str]] = None,
) -> list[RemoteMemberDTO]:
raise OrgSyncProviderError(msg='WeChat Work provider not implemented')
async def test_connection(self) -> dict:
raise OrgSyncProviderError(msg='WeChat Work provider not implemented')
@@ -0,0 +1,121 @@
"""Request/response DTOs for org sync API endpoints."""
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, field_validator
# ---------------------------------------------------------------------------
# Request DTOs
# ---------------------------------------------------------------------------
class OrgSyncConfigCreate(BaseModel):
provider: str
config_name: str
auth_type: str
auth_config: dict
sync_scope: Optional[dict] = None
schedule_type: str = 'manual'
cron_expression: Optional[str] = None
@field_validator('provider')
@classmethod
def validate_provider(cls, v: str) -> str:
allowed = {'feishu', 'wecom', 'dingtalk', 'generic_api'}
if v not in allowed:
raise ValueError(f'provider must be one of {allowed}')
return v
@field_validator('auth_type')
@classmethod
def validate_auth_type(cls, v: str) -> str:
allowed = {'api_key', 'password'}
if v not in allowed:
raise ValueError(f'auth_type must be one of {allowed}')
return v
@field_validator('schedule_type')
@classmethod
def validate_schedule_type(cls, v: str) -> str:
allowed = {'manual', 'cron'}
if v not in allowed:
raise ValueError(f'schedule_type must be one of {allowed}')
return v
class OrgSyncConfigUpdate(BaseModel):
auth_type: Optional[str] = None
auth_config: Optional[dict] = None
sync_scope: Optional[dict] = None
schedule_type: Optional[str] = None
cron_expression: Optional[str] = None
status: Optional[str] = None
config_name: Optional[str] = None
# ---------------------------------------------------------------------------
# Response DTOs
# ---------------------------------------------------------------------------
class OrgSyncConfigRead(BaseModel):
id: int
provider: str
config_name: str
auth_type: str
auth_config: dict # masked
sync_scope: Optional[dict] = None
schedule_type: str
cron_expression: Optional[str] = None
sync_status: str
last_sync_at: Optional[datetime] = None
last_sync_result: Optional[str] = None
status: str
create_user: Optional[int] = None
create_time: Optional[datetime] = None
update_time: Optional[datetime] = None
class OrgSyncLogRead(BaseModel):
id: int
config_id: int
trigger_type: str
trigger_user: Optional[int] = None
status: str
dept_created: int = 0
dept_updated: int = 0
dept_archived: int = 0
member_created: int = 0
member_updated: int = 0
member_disabled: int = 0
member_reactivated: int = 0
error_details: Optional[list] = None
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
create_time: Optional[datetime] = None
class RemoteTreeNode(BaseModel):
external_id: str
name: str
children: list['RemoteTreeNode'] = []
# ---------------------------------------------------------------------------
# Sensitive field masking (AC-34)
# ---------------------------------------------------------------------------
SENSITIVE_KEYS = {'app_secret', 'api_key', 'password', 'secret', 'token'}
def mask_sensitive_fields(auth_config: dict) -> dict:
"""Replace sensitive values with '****' for API responses."""
masked = {}
for key, value in auth_config.items():
if key.lower() in SENSITIVE_KEYS or 'secret' in key.lower() or 'password' in key.lower():
masked[key] = '****'
elif isinstance(value, dict):
masked[key] = mask_sensitive_fields(value)
else:
masked[key] = value
return masked
@@ -0,0 +1,25 @@
"""Standard DTOs for data fetched from third-party org providers."""
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class RemoteDepartmentDTO:
"""A department fetched from a third-party provider."""
external_id: str
name: str
parent_external_id: Optional[str] = None # None = root
sort_order: int = 0
@dataclass
class RemoteMemberDTO:
"""An employee fetched from a third-party provider."""
external_id: str
name: str
email: Optional[str] = None
phone: Optional[str] = None
primary_dept_external_id: str = ''
secondary_dept_external_ids: list[str] = field(default_factory=list)
status: str = 'active' # active / disabled
@@ -0,0 +1,596 @@
"""OrgSyncService — sync orchestrator.
Implements the 16-step flow described in spec §7:
load config → acquire lock → create log → authenticate → fetch →
reconcile → apply → update log → release lock.
Bypasses DepartmentService permission checks (AD-11) by directly
operating DAO + DepartmentChangeHandler for system-level sync.
"""
import secrets
from datetime import datetime
from typing import Optional
from loguru import logger
from bisheng.common.errcode.org_sync import (
OrgSyncAlreadyRunningError,
OrgSyncConfigDisabledError,
OrgSyncConfigNotFoundError,
)
from bisheng.database.models.department import (
Department,
DepartmentDao,
UserDepartment,
UserDepartmentDao,
)
from bisheng.database.models.tenant import UserTenant, UserTenantDao
from bisheng.department.domain.services.department_change_handler import DepartmentChangeHandler
from bisheng.org_sync.domain.models.org_sync import (
OrgSyncConfig,
OrgSyncConfigDao,
OrgSyncLog,
OrgSyncLogDao,
decrypt_auth_config,
)
from bisheng.org_sync.domain.providers.base import get_provider
from bisheng.org_sync.domain.services.reconciler import (
ArchiveDept,
CreateDept,
CreateMember,
DisableMember,
MoveDept,
ReactivateMember,
TransferMember,
UpdateDept,
UpdateMember,
reconcile_departments,
reconcile_members,
)
from bisheng.user.domain.models.user import User, UserDao
REDIS_LOCK_KEY_PREFIX = 'bisheng:lock:org_sync:'
REDIS_LOCK_TTL = 1800 # 30 minutes
class OrgSyncService:
@classmethod
async def execute_sync(
cls,
config_id: int,
trigger_type: str,
trigger_user: Optional[int] = None,
) -> int:
"""Main orchestration flow. Returns log_id."""
# Step 1: Load config
config = await OrgSyncConfigDao.aget_by_id(config_id)
if not config:
raise OrgSyncConfigNotFoundError()
if config.status == 'disabled':
raise OrgSyncConfigDisabledError()
# Step 2: Acquire mutex (DB CAS + Redis lock)
acquired = await OrgSyncConfigDao.aset_sync_status(
config_id, 'idle', 'running',
)
if not acquired:
raise OrgSyncAlreadyRunningError()
redis_lock = await cls._acquire_redis_lock(config_id)
# Step 3: Create log entry
log = OrgSyncLog(
tenant_id=config.tenant_id,
config_id=config_id,
trigger_type=trigger_type,
trigger_user=trigger_user,
status='running',
start_time=datetime.now(),
)
log = await OrgSyncLogDao.acreate(log)
log_id = log.id
errors: list[dict] = []
stats = {
'dept_created': 0, 'dept_updated': 0, 'dept_archived': 0,
'member_created': 0, 'member_updated': 0, 'member_disabled': 0,
'member_reactivated': 0,
}
try:
# Step 4: Decrypt auth_config and instantiate provider
auth_config = decrypt_auth_config(config.auth_config)
provider = get_provider(config.provider, auth_config)
# Step 5: Authenticate
await provider.authenticate()
# Step 6: Fetch remote departments
scope = config.sync_scope
root_dept_ids = scope.get('root_dept_ids') if scope else None
remote_depts = await provider.fetch_departments(root_dept_ids)
# Step 7: Load local departments
local_depts = await DepartmentDao.aget_all_active()
local_depts = [
d for d in local_depts if d.tenant_id == config.tenant_id
]
# Step 8: Reconcile departments
dept_ops = reconcile_departments(
remote_depts, local_depts, config.provider,
)
# Step 9: Apply department operations
ext_to_local = await cls._apply_dept_ops(
dept_ops, config, stats, errors,
)
# Step 10: Fetch remote members
dept_ext_ids = [d.external_id for d in remote_depts]
remote_members = await provider.fetch_members(dept_ext_ids)
# Step 11: Load local users
local_users = await UserDao.aget_by_source(
config.provider, config.tenant_id,
)
# Build user department map
local_user_depts: dict[int, list[UserDepartment]] = {}
for user in local_users:
from bisheng.database.models.department import UserDepartmentDao
depts = await UserDepartmentDao.aget_user_departments(user.user_id)
local_user_depts[user.user_id] = depts
# Step 12: Reconcile members
member_ops = reconcile_members(
remote_members, local_users, local_user_depts,
ext_to_local, config.provider,
)
# Step 13: Apply member operations
await cls._apply_member_ops(
member_ops, config, stats, errors, ext_to_local,
)
except Exception as e:
logger.exception(f'Sync failed for config {config_id}: {e}')
errors.append({
'entity_type': 'system',
'external_id': '',
'error_msg': str(e),
})
# Step 14: Update log
final_status = 'success'
if errors:
final_status = 'partial' if any(
stats[k] > 0 for k in stats
) else 'failed'
log.status = final_status
log.dept_created = stats['dept_created']
log.dept_updated = stats['dept_updated']
log.dept_archived = stats['dept_archived']
log.member_created = stats['member_created']
log.member_updated = stats['member_updated']
log.member_disabled = stats['member_disabled']
log.member_reactivated = stats['member_reactivated']
log.error_details = errors or None
log.end_time = datetime.now()
await OrgSyncLogDao.aupdate(log)
# Step 15: Update config
config.last_sync_at = datetime.now()
config.last_sync_result = final_status
await OrgSyncConfigDao.aupdate(config)
# Step 16: Release lock
await OrgSyncConfigDao.aset_sync_status(config_id, 'running', 'idle')
await cls._release_redis_lock(config_id, redis_lock)
return log_id
# ------------------------------------------------------------------
# Department operations
# ------------------------------------------------------------------
@classmethod
async def _apply_dept_ops(
cls,
ops: list,
config: OrgSyncConfig,
stats: dict,
errors: list[dict],
) -> dict[str, int]:
"""Execute department operations. Returns ext_id → local dept.id map."""
# Build initial mapping from existing local departments
local_depts = await DepartmentDao.aget_all_active()
local_depts = [d for d in local_depts if d.tenant_id == config.tenant_id]
ext_to_local: dict[str, int] = {}
for d in local_depts:
if d.external_id:
ext_to_local[d.external_id] = d.id
for op in ops:
try:
if isinstance(op, CreateDept):
await cls._create_dept(op, config, ext_to_local)
stats['dept_created'] += 1
elif isinstance(op, UpdateDept):
await cls._update_dept(op, config)
stats['dept_updated'] += 1
elif isinstance(op, MoveDept):
await cls._move_dept(op, ext_to_local)
stats['dept_updated'] += 1
elif isinstance(op, ArchiveDept):
await cls._archive_dept(op)
stats['dept_archived'] += 1
except Exception as e:
ext_id = ''
if isinstance(op, CreateDept):
ext_id = op.remote.external_id
elif hasattr(op, 'local'):
ext_id = op.local.external_id or ''
logger.warning(f'Dept op failed ({type(op).__name__}): {e}')
errors.append({
'entity_type': 'department',
'external_id': ext_id,
'error_msg': str(e),
})
return ext_to_local
@classmethod
async def _create_dept(
cls, op: CreateDept, config: OrgSyncConfig,
ext_to_local: dict[str, int],
) -> None:
# Resolve parent
parent_id: Optional[int] = None
if op.remote.parent_external_id:
parent_id = ext_to_local.get(op.remote.parent_external_id)
if parent_id is None:
# Try to find by external_id in DB
from bisheng.core.database import get_async_db_session
from sqlmodel import select
async with get_async_db_session() as session:
result = await session.exec(
select(Department).where(
Department.external_id == op.remote.parent_external_id,
Department.tenant_id == config.tenant_id,
)
)
parent = result.first()
if parent:
parent_id = parent.id
else:
# Root department — find tenant root
root = await DepartmentDao.aget_root_by_tenant(config.tenant_id)
if root:
parent_id = root.id
# Build path
path = '/'
if parent_id:
parent_dept = await DepartmentDao.aget_by_id(parent_id)
if parent_dept:
path = parent_dept.path
# Generate business key
import uuid
dept_id = f'BS@{uuid.uuid4().hex[:5]}'
dept = Department(
dept_id=dept_id,
name=op.remote.name,
parent_id=parent_id,
tenant_id=config.tenant_id,
path=path, # temporary, will be updated
sort_order=op.remote.sort_order,
source=config.provider,
external_id=op.remote.external_id,
)
dept = await DepartmentDao.acreate(dept)
# Fix path with actual ID
dept.path = f'{path}{dept.id}/'
await DepartmentDao.aupdate(dept)
# Update mapping
ext_to_local[op.remote.external_id] = dept.id
# OpenFGA tuple via ChangeHandler
if parent_id:
tuple_ops = DepartmentChangeHandler.on_created(dept.id, parent_id)
await DepartmentChangeHandler.execute_async(tuple_ops)
@classmethod
async def _update_dept(cls, op: UpdateDept, config: OrgSyncConfig) -> None:
op.local.name = op.new_name
if op.change_source:
op.local.source = config.provider
await DepartmentDao.aupdate(op.local)
@classmethod
async def _move_dept(
cls, op: MoveDept, ext_to_local: dict[str, int],
) -> None:
old_parent_id = op.local.parent_id
new_parent_id = ext_to_local.get(op.new_parent_external_id) if op.new_parent_external_id else None
if new_parent_id is None or new_parent_id == old_parent_id:
return
# Update parent_id
op.local.parent_id = new_parent_id
# Rebuild path
new_parent = await DepartmentDao.aget_by_id(new_parent_id)
old_path = op.local.path
new_prefix = f'{new_parent.path}{op.local.id}/' if new_parent else f'/{op.local.id}/'
op.local.path = new_prefix
await DepartmentDao.aupdate(op.local)
# Update all descendant paths
if old_path and old_path != new_prefix:
await DepartmentDao.aupdate_paths_batch(old_path, new_prefix)
# OpenFGA tuples
if old_parent_id:
tuple_ops = DepartmentChangeHandler.on_moved(
op.local.id, old_parent_id, new_parent_id,
)
await DepartmentChangeHandler.execute_async(tuple_ops)
@classmethod
async def _archive_dept(cls, op: ArchiveDept) -> None:
op.local.status = 'archived'
await DepartmentDao.aupdate(op.local)
# OpenFGA cleanup
if op.local.parent_id:
tuple_ops = DepartmentChangeHandler.on_archived(
op.local.id, op.local.parent_id,
)
await DepartmentChangeHandler.execute_async(tuple_ops)
# ------------------------------------------------------------------
# Member operations
# ------------------------------------------------------------------
@classmethod
async def _apply_member_ops(
cls,
ops: list,
config: OrgSyncConfig,
stats: dict,
errors: list[dict],
ext_to_local_dept: dict[str, int],
) -> None:
for op in ops:
try:
if isinstance(op, CreateMember):
await cls._create_member(op, config, ext_to_local_dept)
stats['member_created'] += 1
elif isinstance(op, UpdateMember):
await cls._update_member(op, config)
stats['member_updated'] += 1
elif isinstance(op, TransferMember):
await cls._transfer_member(op, config, ext_to_local_dept)
stats['member_updated'] += 1
elif isinstance(op, DisableMember):
await cls._disable_member(op)
stats['member_disabled'] += 1
elif isinstance(op, ReactivateMember):
await cls._reactivate_member(op, config, ext_to_local_dept)
stats['member_reactivated'] += 1
except Exception as e:
ext_id = ''
if isinstance(op, CreateMember):
ext_id = op.remote.external_id
elif isinstance(op, ReactivateMember):
ext_id = op.remote.external_id
logger.warning(f'Member op failed ({type(op).__name__}): {e}')
errors.append({
'entity_type': 'member',
'external_id': ext_id,
'error_msg': str(e),
})
@classmethod
async def _create_member(
cls, op: CreateMember, config: OrgSyncConfig,
ext_to_local_dept: dict[str, int],
) -> None:
# Create User with random password (AD-07)
password_hash = secrets.token_hex(32)
user = User(
user_name=op.remote.name,
email=op.remote.email,
phone_number=op.remote.phone,
source=config.provider,
external_id=op.remote.external_id,
password=password_hash,
)
user = await UserDao.add_user_and_default_role(user)
# Create UserTenant
from bisheng.core.database import get_async_db_session
async with get_async_db_session() as session:
ut = UserTenant(user_id=user.user_id, tenant_id=config.tenant_id)
session.add(ut)
await session.commit()
# Create UserDepartment entries
all_dept_ids: list[int] = []
primary_dept_id = ext_to_local_dept.get(op.remote.primary_dept_external_id)
if primary_dept_id:
await UserDepartmentDao.aadd_member(
user.user_id, primary_dept_id, is_primary=1, source=config.provider,
)
all_dept_ids.append(primary_dept_id)
for ext_id in op.remote.secondary_dept_external_ids:
dept_id = ext_to_local_dept.get(ext_id)
if dept_id and dept_id != primary_dept_id:
await UserDepartmentDao.aadd_member(
user.user_id, dept_id, is_primary=0, source=config.provider,
)
all_dept_ids.append(dept_id)
# OpenFGA member tuples
for dept_id in all_dept_ids:
tuple_ops = DepartmentChangeHandler.on_members_added(
dept_id, [user.user_id],
)
await DepartmentChangeHandler.execute_async(tuple_ops)
@classmethod
async def _update_member(
cls, op: UpdateMember, config: OrgSyncConfig,
) -> None:
user = await UserDao.aget_user(op.user_id)
if not user:
return
if op.new_name:
user.user_name = op.new_name
if op.new_email is not None:
user.email = op.new_email
if op.new_phone is not None:
user.phone_number = op.new_phone
if op.change_source:
user.source = config.provider
await UserDao.aupdate_user(user)
@classmethod
async def _transfer_member(
cls, op: TransferMember, config: OrgSyncConfig,
ext_to_local_dept: dict[str, int],
) -> None:
# Primary department change
if op.old_primary_dept_id is not None:
new_primary_id = ext_to_local_dept.get(op.new_primary_dept_external_id)
if new_primary_id:
# Remove old primary membership
await UserDepartmentDao.aremove_member(op.user_id, op.old_primary_dept_id)
tuple_ops = DepartmentChangeHandler.on_member_removed(
op.old_primary_dept_id, op.user_id,
)
await DepartmentChangeHandler.execute_async(tuple_ops)
# Add new primary membership
await UserDepartmentDao.aadd_member(
op.user_id, new_primary_id, is_primary=1, source=config.provider,
)
tuple_ops = DepartmentChangeHandler.on_members_added(
new_primary_id, [op.user_id],
)
await DepartmentChangeHandler.execute_async(tuple_ops)
# Add new secondary departments
for ext_id in op.add_secondary_external_ids:
dept_id = ext_to_local_dept.get(ext_id)
if dept_id:
await UserDepartmentDao.aadd_member(
op.user_id, dept_id, is_primary=0, source=config.provider,
)
tuple_ops = DepartmentChangeHandler.on_members_added(
dept_id, [op.user_id],
)
await DepartmentChangeHandler.execute_async(tuple_ops)
# Remove old secondary departments
for dept_id in op.remove_secondary_dept_ids:
await UserDepartmentDao.aremove_member(op.user_id, dept_id)
tuple_ops = DepartmentChangeHandler.on_member_removed(dept_id, op.user_id)
await DepartmentChangeHandler.execute_async(tuple_ops)
@classmethod
async def _disable_member(cls, op: DisableMember) -> None:
user = await UserDao.aget_user(op.user_id)
if not user:
return
user.delete = 1
await UserDao.aupdate_user(user)
# Clean up all department memberships + OpenFGA tuples
for dept_id in op.dept_ids:
await UserDepartmentDao.aremove_member(op.user_id, dept_id)
tuple_ops = DepartmentChangeHandler.on_member_removed(dept_id, op.user_id)
await DepartmentChangeHandler.execute_async(tuple_ops)
@classmethod
async def _reactivate_member(
cls, op: ReactivateMember, config: OrgSyncConfig,
ext_to_local_dept: dict[str, int],
) -> None:
user = await UserDao.aget_user(op.user_id)
if not user:
return
user.delete = 0
user.source = config.provider
await UserDao.aupdate_user(user)
# Rebuild department memberships
all_dept_ids: list[int] = []
primary_dept_id = ext_to_local_dept.get(op.remote.primary_dept_external_id)
if primary_dept_id:
await UserDepartmentDao.aadd_member(
op.user_id, primary_dept_id, is_primary=1, source=config.provider,
)
all_dept_ids.append(primary_dept_id)
for ext_id in op.remote.secondary_dept_external_ids:
dept_id = ext_to_local_dept.get(ext_id)
if dept_id and dept_id != primary_dept_id:
await UserDepartmentDao.aadd_member(
op.user_id, dept_id, is_primary=0, source=config.provider,
)
all_dept_ids.append(dept_id)
for dept_id in all_dept_ids:
tuple_ops = DepartmentChangeHandler.on_members_added(
dept_id, [op.user_id],
)
await DepartmentChangeHandler.execute_async(tuple_ops)
# ------------------------------------------------------------------
# Redis lock helpers
# ------------------------------------------------------------------
@classmethod
async def _acquire_redis_lock(cls, config_id: int) -> Optional[str]:
"""Acquire Redis distributed lock. Returns lock token or None."""
try:
from bisheng.core.cache.redis_manager import get_redis_client
redis = await get_redis_client()
lock_key = f'{REDIS_LOCK_KEY_PREFIX}{config_id}'
token = secrets.token_hex(16)
acquired = await redis.async_connection.set(
lock_key, token, nx=True, ex=REDIS_LOCK_TTL,
)
return token if acquired else None
except Exception as e:
logger.warning(f'Redis lock acquisition failed (non-fatal): {e}')
return None
@classmethod
async def _release_redis_lock(
cls, config_id: int, token: Optional[str],
) -> None:
"""Release Redis lock if we hold it."""
if not token:
return
try:
from bisheng.core.cache.redis_manager import get_redis_client
redis = await get_redis_client()
lock_key = f'{REDIS_LOCK_KEY_PREFIX}{config_id}'
# Only release if we still hold the lock
current = await redis.async_connection.get(lock_key)
if current and current.decode() == token:
await redis.async_connection.delete(lock_key)
except Exception as e:
logger.warning(f'Redis lock release failed (non-fatal): {e}')
@@ -0,0 +1,410 @@
"""Reconciler — pure-logic diff engine for org sync.
Compares remote DTOs against local database records and produces
typed operation lists. No IO, no side effects — fully unit-testable.
"""
from dataclasses import dataclass, field
from typing import Optional
from bisheng.database.models.department import Department, UserDepartment
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
# ---------------------------------------------------------------------------
# Department Operations
# ---------------------------------------------------------------------------
@dataclass
class CreateDept:
remote: RemoteDepartmentDTO
@dataclass
class UpdateDept:
local: Department
new_name: str
change_source: bool = False # True if source was 'local', force-overwritten
@dataclass
class MoveDept:
local: Department
new_parent_external_id: Optional[str]
@dataclass
class ArchiveDept:
local: Department
DeptOperation = CreateDept | UpdateDept | MoveDept | ArchiveDept
# ---------------------------------------------------------------------------
# Member Operations
# ---------------------------------------------------------------------------
@dataclass
class CreateMember:
remote: RemoteMemberDTO
@dataclass
class UpdateMember:
user_id: int
new_name: Optional[str] = None
new_email: Optional[str] = None
new_phone: Optional[str] = None
change_source: bool = False
@dataclass
class TransferMember:
user_id: int
new_primary_dept_external_id: str
old_primary_dept_id: Optional[int] = None
add_secondary_external_ids: list[str] = field(default_factory=list)
remove_secondary_dept_ids: list[int] = field(default_factory=list)
@dataclass
class DisableMember:
user_id: int
dept_ids: list[int] = field(default_factory=list) # departments to clean up
@dataclass
class ReactivateMember:
user_id: int
remote: RemoteMemberDTO
MemberOperation = CreateMember | UpdateMember | TransferMember | DisableMember | ReactivateMember
# ---------------------------------------------------------------------------
# Department Reconciliation
# ---------------------------------------------------------------------------
def reconcile_departments(
remote_depts: list[RemoteDepartmentDTO],
local_depts: list[Department],
source: str,
) -> list[DeptOperation]:
"""Compare remote departments against local and produce operations.
Args:
remote_depts: Departments fetched from the provider.
local_depts: Active local departments (all sources within tenant).
source: Provider source string (e.g. 'feishu').
Returns:
Topologically sorted list of DeptOperations
(creates: parent-first, archives: child-first).
"""
# Build lookup maps
remote_map: dict[str, RemoteDepartmentDTO] = {
d.external_id: d for d in remote_depts
}
# Local departments keyed by external_id (only those with an external_id)
local_by_ext: dict[str, Department] = {}
for d in local_depts:
if d.external_id:
local_by_ext[d.external_id] = d
creates: list[CreateDept] = []
updates: list[UpdateDept] = []
moves: list[MoveDept] = []
archives: list[ArchiveDept] = []
# Pass 1: detect creates, updates, moves from remote data
for ext_id, remote in remote_map.items():
local = local_by_ext.get(ext_id)
if local is None:
creates.append(CreateDept(remote=remote))
else:
# Name change
if remote.name != local.name:
change_src = (local.source == 'local')
updates.append(UpdateDept(
local=local,
new_name=remote.name,
change_source=change_src,
))
elif local.source == 'local':
# Source mismatch but no name change — still adopt
updates.append(UpdateDept(
local=local,
new_name=remote.name,
change_source=True,
))
# Parent change
current_parent_ext = _get_parent_external_id(local, local_depts)
if remote.parent_external_id != current_parent_ext:
moves.append(MoveDept(
local=local,
new_parent_external_id=remote.parent_external_id,
))
# Pass 2: detect archives (local exists with matching source, but not in remote)
for ext_id, local in local_by_ext.items():
if local.source == source and ext_id not in remote_map:
archives.append(ArchiveDept(local=local))
# Also archive child departments of archived departments
archived_ids = {a.local.id for a in archives}
for dept in local_depts:
if dept.id not in archived_ids and dept.status == 'active':
if _is_descendant_of_any(dept, archived_ids, local_depts):
archives.append(ArchiveDept(local=dept))
archived_ids.add(dept.id)
# Topological sort: creates parent-first
creates = _topo_sort_creates(creates, remote_map)
# Archives: child-first (reverse of parent-first order by path depth)
archives.sort(key=lambda a: a.local.path.count('/'), reverse=True)
# Combine in execution order: creates → updates → moves → archives
result: list[DeptOperation] = []
result.extend(creates)
result.extend(updates)
result.extend(moves)
result.extend(archives)
return result
def _get_parent_external_id(
dept: Department, all_depts: list[Department],
) -> Optional[str]:
"""Get the external_id of a department's parent."""
if dept.parent_id is None:
return None
for d in all_depts:
if d.id == dept.parent_id:
return d.external_id
return None
def _is_descendant_of_any(
dept: Department, ancestor_ids: set[int], all_depts: list[Department],
) -> bool:
"""Check if dept is a descendant of any department in ancestor_ids."""
parent_map = {d.id: d.parent_id for d in all_depts}
current = dept.parent_id
visited: set[int] = set()
while current is not None:
if current in ancestor_ids:
return True
if current in visited:
break # cycle detected
visited.add(current)
current = parent_map.get(current)
return False
def _topo_sort_creates(
creates: list[CreateDept],
remote_map: dict[str, RemoteDepartmentDTO],
) -> list[CreateDept]:
"""Sort create operations so parents come before children.
Uses Kahn's algorithm. Detects cycles and drops affected nodes.
"""
if not creates:
return []
ext_ids = {c.remote.external_id for c in creates}
create_map = {c.remote.external_id: c for c in creates}
# Build in-degree (dependency = parent must be created first)
in_degree: dict[str, int] = {eid: 0 for eid in ext_ids}
children: dict[str, list[str]] = {eid: [] for eid in ext_ids}
for c in creates:
parent_ext = c.remote.parent_external_id
if parent_ext and parent_ext in ext_ids:
in_degree[c.remote.external_id] += 1
children[parent_ext].append(c.remote.external_id)
# Kahn's algorithm
queue = [eid for eid, deg in in_degree.items() if deg == 0]
sorted_result: list[CreateDept] = []
while queue:
eid = queue.pop(0)
sorted_result.append(create_map[eid])
for child_eid in children.get(eid, []):
in_degree[child_eid] -= 1
if in_degree[child_eid] == 0:
queue.append(child_eid)
# If some nodes remain (cycle), skip them
return sorted_result
# ---------------------------------------------------------------------------
# Member Reconciliation
# ---------------------------------------------------------------------------
def reconcile_members(
remote_members: list[RemoteMemberDTO],
local_users: list, # List[User] — typed loosely to avoid circular import
local_user_depts: dict[int, list[UserDepartment]],
ext_to_local_dept: dict[str, int],
source: str,
) -> list[MemberOperation]:
"""Compare remote members against local users and produce operations.
Args:
remote_members: Members fetched from the provider.
local_users: Local users (filtered by source + tenant).
local_user_depts: user_id → list of UserDepartment records.
ext_to_local_dept: external_id → local department.id mapping.
source: Provider source string.
Returns:
List of MemberOperations in execution order.
"""
remote_map: dict[str, RemoteMemberDTO] = {
m.external_id: m for m in remote_members
}
local_by_ext: dict[str, object] = {} # external_id → User
for u in local_users:
if u.external_id:
local_by_ext[u.external_id] = u
creates: list[CreateMember] = []
updates: list[UpdateMember] = []
transfers: list[TransferMember] = []
disables: list[DisableMember] = []
reactivates: list[ReactivateMember] = []
# Pass 1: process remote members
for ext_id, remote in remote_map.items():
local = local_by_ext.get(ext_id)
if local is None:
if remote.status == 'active':
creates.append(CreateMember(remote=remote))
continue
# Reactivate if previously disabled
if getattr(local, 'delete', 0) == 1 and remote.status == 'active':
reactivates.append(ReactivateMember(
user_id=local.user_id,
remote=remote,
))
continue
# Disable if remote says disabled
if remote.status == 'disabled':
user_depts = local_user_depts.get(local.user_id, [])
disables.append(DisableMember(
user_id=local.user_id,
dept_ids=[ud.department_id for ud in user_depts],
))
continue
# Check info changes
change_source = (getattr(local, 'source', '') == 'local')
needs_update = False
new_name = None
new_email = None
new_phone = None
if remote.name and remote.name != getattr(local, 'user_name', ''):
new_name = remote.name
needs_update = True
if remote.email is not None and remote.email != getattr(local, 'email', ''):
new_email = remote.email
needs_update = True
if remote.phone is not None and remote.phone != getattr(local, 'phone_number', ''):
new_phone = remote.phone
needs_update = True
if needs_update or change_source:
updates.append(UpdateMember(
user_id=local.user_id,
new_name=new_name,
new_email=new_email,
new_phone=new_phone,
change_source=change_source,
))
# Check department changes
user_depts = local_user_depts.get(local.user_id, [])
_check_dept_changes(
local, remote, user_depts, ext_to_local_dept, transfers,
)
# Pass 2: detect disables (local exists with matching source, not in remote)
for ext_id, local in local_by_ext.items():
if (
getattr(local, 'source', '') == source
and ext_id not in remote_map
and getattr(local, 'delete', 0) == 0
):
user_depts = local_user_depts.get(local.user_id, [])
disables.append(DisableMember(
user_id=local.user_id,
dept_ids=[ud.department_id for ud in user_depts],
))
result: list[MemberOperation] = []
result.extend(creates)
result.extend(updates)
result.extend(transfers)
result.extend(disables)
result.extend(reactivates)
return result
def _check_dept_changes(
local_user,
remote: RemoteMemberDTO,
user_depts: list[UserDepartment],
ext_to_local_dept: dict[str, int],
transfers: list[TransferMember],
) -> None:
"""Detect primary/secondary department changes for an existing user."""
current_primary_dept_id: Optional[int] = None
current_secondary_dept_ids: set[int] = set()
for ud in user_depts:
if ud.is_primary == 1:
current_primary_dept_id = ud.department_id
else:
current_secondary_dept_ids.add(ud.department_id)
# Desired state from remote
new_primary_dept_id = ext_to_local_dept.get(remote.primary_dept_external_id)
new_secondary_dept_ids: set[int] = set()
add_secondary_ext: list[str] = []
for ext_id in remote.secondary_dept_external_ids:
local_id = ext_to_local_dept.get(ext_id)
if local_id is not None:
new_secondary_dept_ids.add(local_id)
primary_changed = (
new_primary_dept_id is not None
and new_primary_dept_id != current_primary_dept_id
)
to_add_secondary = new_secondary_dept_ids - current_secondary_dept_ids
to_remove_secondary = current_secondary_dept_ids - new_secondary_dept_ids
if primary_changed or to_add_secondary or to_remove_secondary:
# Build add_secondary_external_ids from to_add_secondary
dept_id_to_ext = {v: k for k, v in ext_to_local_dept.items()}
add_ext = [dept_id_to_ext[did] for did in to_add_secondary if did in dept_id_to_ext]
transfers.append(TransferMember(
user_id=local_user.user_id,
new_primary_dept_external_id=remote.primary_dept_external_id,
old_primary_dept_id=current_primary_dept_id if primary_changed else None,
add_secondary_external_ids=add_ext,
remove_secondary_dept_ids=list(to_remove_secondary),
))
+44 -1
View File
@@ -2,7 +2,7 @@ from datetime import datetime
from typing import List, Optional
from pydantic import field_validator
from sqlalchemy import Column, DateTime, func, text
from sqlalchemy import Column, DateTime, String, UniqueConstraint, func, text
from sqlalchemy.orm import selectinload
from sqlmodel import Field, select, Relationship, col
@@ -22,6 +22,21 @@ class UserBase(SQLModelSerializable):
dept_id: Optional[str] = Field(default=None, index=True)
remark: Optional[str] = Field(default=None, index=False)
avatar: Optional[str] = Field(default=None, index=False)
source: str = Field(
default='local',
sa_column=Column(
String(32), nullable=False,
server_default=text("'local'"),
comment='Source: local/feishu/wecom/dingtalk/generic_api',
),
)
external_id: Optional[str] = Field(
default=None,
sa_column=Column(
String(128), nullable=True,
comment='External employee ID for sync',
),
)
delete: int = Field(default=0, index=False)
create_time: Optional[datetime] = Field(default=None, sa_column=Column(
DateTime, nullable=False, index=True, server_default=text('CURRENT_TIMESTAMP')))
@@ -48,6 +63,9 @@ class User(UserBase, table=True):
roles: List["Role"] = Relationship(link_model=UserRole)
__tablename__ = "user"
__table_args__ = (
UniqueConstraint('source', 'external_id', name='uk_user_source_external_id'),
)
class UserRead(UserBase):
@@ -280,3 +298,28 @@ class UserDao(UserBase):
statement = select(User).order_by(col(User.user_id).asc()).limit(1)
with get_sync_db_session() as session:
return session.exec(statement).first()
@classmethod
async def aget_by_source_external_id(cls, source: str, external_id: str) -> Optional['User']:
"""Get user by source + external_id combination (for org sync matching)."""
async with get_async_db_session() as session:
statement = select(User).where(
User.source == source,
User.external_id == external_id,
)
result = await session.exec(statement)
return result.first()
@classmethod
async def aget_by_source(cls, source: str, tenant_id: int) -> List['User']:
"""Get all users from a given source within a tenant (for reconcile)."""
from bisheng.database.models.user_tenant import UserTenant
async with get_async_db_session() as session:
statement = select(User).join(
UserTenant, User.user_id == UserTenant.user_id,
).where(
User.source == source,
UserTenant.tenant_id == tenant_id,
)
result = await session.exec(statement)
return result.all()
+1
View File
@@ -8,3 +8,4 @@ from bisheng.worker.telemetry.mid_table import sync_mid_user_increment, sync_mid
from bisheng.worker.test.test import add
from bisheng.worker.workflow.tasks import execute_workflow, continue_workflow, stop_workflow
from bisheng.worker.permission.retry_failed_tuples import retry_failed_tuples
from bisheng.worker.org_sync.tasks import execute_org_sync, check_org_sync_schedules
@@ -0,0 +1,88 @@
"""Celery tasks for org sync (F009).
- execute_org_sync: async sync execution, dispatched by API or beat.
- check_org_sync_schedules: beat task, checks active cron configs every 60s.
Tenant context: propagated via Celery headers (INV-8), automatically restored
by the before_task signal in bisheng.worker.tenant_context.
"""
import asyncio
import logging
from datetime import datetime
from bisheng.worker.main import bisheng_celery
logger = logging.getLogger(__name__)
@bisheng_celery.task(acks_late=True, time_limit=1800, soft_time_limit=1500)
def execute_org_sync(config_id: int, trigger_type: str, trigger_user: int = None):
"""Execute org sync for a given config. Runs in knowledge_celery queue."""
loop = asyncio.new_event_loop()
try:
loop.run_until_complete(_execute_org_sync_async(
config_id, trigger_type, trigger_user,
))
finally:
loop.close()
async def _execute_org_sync_async(
config_id: int, trigger_type: str, trigger_user: int = None,
):
from bisheng.org_sync.domain.services.org_sync_service import OrgSyncService
try:
log_id = await OrgSyncService.execute_sync(
config_id, trigger_type, trigger_user,
)
logger.info(f'Org sync completed for config {config_id}, log_id={log_id}')
except Exception:
logger.exception(f'Org sync failed for config {config_id}')
@bisheng_celery.task(acks_late=True)
def check_org_sync_schedules():
"""Beat task: check active cron configs and dispatch if due.
Runs every 60s. Uses croniter to evaluate cron expressions.
"""
loop = asyncio.new_event_loop()
try:
loop.run_until_complete(_check_schedules_async())
finally:
loop.close()
async def _check_schedules_async():
from bisheng.org_sync.domain.models.org_sync import OrgSyncConfigDao
try:
from croniter import croniter
except ImportError:
logger.error('croniter not installed, org sync cron scheduling disabled')
return
configs = await OrgSyncConfigDao.aget_active_cron_configs()
now = datetime.now()
for config in configs:
if not config.cron_expression:
continue
try:
cron = croniter(config.cron_expression, config.last_sync_at or now)
next_run = cron.get_next(datetime)
if next_run <= now:
# Due for execution — dispatch
logger.info(
f'Dispatching scheduled sync for config {config.id} '
f'(cron={config.cron_expression})',
)
execute_org_sync.apply_async(
args=[config.id, 'scheduled', None],
queue='knowledge_celery',
)
except Exception as e:
logger.warning(
f'Cron evaluation failed for config {config.id}: {e}',
)
+246
View File
@@ -0,0 +1,246 @@
"""
E2E tests for F009: Org Sync
Prerequisites:
- Backend running on localhost:7860 (or E2E_API_BASE env var)
- Default admin account: admin/Bisheng@top1 (or E2E_ADMIN_PASSWORD env var)
- Default tenant (id=1) with root department
Covers:
- AC-01: Create config (POST /org-sync/configs)
- AC-02: Duplicate config error (22001)
- AC-03: List configs (GET /org-sync/configs)
- AC-04: Get config detail (GET /org-sync/configs/{id})
- AC-05: Update config (PUT /org-sync/configs/{id})
- AC-06: Merge auth_config on update
- AC-07: Delete config (soft delete)
- AC-08: Cross-tenant rejection (22000)
- AC-09: Test connection (POST /org-sync/configs/{id}/test) — requires real provider
- AC-10: Test connection auth failure (22002) — requires real provider
- AC-12: Provider not implemented (22004) — WeChat Work
- AC-29: Get sync logs (GET /org-sync/configs/{id}/logs)
- AC-33: Non-admin permission denied (22005)
- AC-34: Auth config masking
Data isolation: All test configs use 'e2e-f009-' prefix in config_name.
"""
import os
import httpx
import pytest
API_BASE = os.environ.get('E2E_API_BASE', 'http://localhost:7860/api/v1')
HEALTH_URL = API_BASE.replace('/api/v1', '') + '/health'
PREFIX = 'e2e-f009-'
# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def _login(client: httpx.Client, username: str = 'admin', password: str = None) -> str:
"""Login and return JWT token."""
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
import base64
pubkey_resp = client.get(f'{API_BASE}/user/public_key')
assert pubkey_resp.status_code == 200
public_key_pem = pubkey_resp.json()['data']['public_key']
if password is None:
password = os.environ.get('E2E_ADMIN_PASSWORD', 'Bisheng@top1')
public_key = serialization.load_pem_public_key(public_key_pem.encode())
encrypted = public_key.encrypt(password.encode(), padding.PKCS1v15())
encrypted_password = base64.b64encode(encrypted).decode()
resp = client.post(f'{API_BASE}/user/login', json={
'user_name': username,
'password': encrypted_password,
})
assert resp.status_code == 200
return resp.cookies.get('access_token_cookie', '')
@pytest.fixture(scope='module')
def admin_client():
"""Authenticated httpx client for admin user."""
client = httpx.Client(base_url=API_BASE, timeout=30)
# Attempt health check
try:
health = client.get(HEALTH_URL)
if health.status_code != 200:
pytest.skip('Backend not running')
except httpx.ConnectError:
pytest.skip('Backend not running')
token = _login(client)
client.cookies.set('access_token_cookie', token)
yield client
client.close()
@pytest.fixture(scope='module')
def config_id(admin_client):
"""Create a test config and return its ID. Cleanup after tests."""
resp = admin_client.post('/org-sync/configs', json={
'provider': 'generic_api',
'config_name': f'{PREFIX}test-config',
'auth_type': 'api_key',
'auth_config': {
'endpoint_url': 'https://httpbin.org/get',
'api_key': 'test-key-12345',
'departments_url': 'https://httpbin.org/json',
'members_url': 'https://httpbin.org/json',
},
'schedule_type': 'manual',
})
assert resp.status_code == 200
data = resp.json()
assert data['status_code'] == 200, f"Create failed: {data}"
cid = data['data']['id']
yield cid
# Cleanup: soft delete
admin_client.delete(f'/org-sync/configs/{cid}')
# ---------------------------------------------------------------------------
# Config CRUD Tests
# ---------------------------------------------------------------------------
class TestConfigCRUD:
def test_create_config(self, admin_client, config_id):
"""AC-01: create returns proper structure."""
resp = admin_client.get(f'/org-sync/configs/{config_id}')
data = resp.json()['data']
assert data['provider'] == 'generic_api'
assert data['config_name'] == f'{PREFIX}test-config'
assert data['status'] == 'active'
assert data['sync_status'] == 'idle'
def test_create_duplicate(self, admin_client):
"""AC-02: duplicate provider+name → 22001."""
resp = admin_client.post('/org-sync/configs', json={
'provider': 'generic_api',
'config_name': f'{PREFIX}test-config',
'auth_type': 'api_key',
'auth_config': {'api_key': 'dup'},
})
data = resp.json()
assert data['status_code'] == 22001
def test_list_configs(self, admin_client, config_id):
"""AC-03: list returns configs for current tenant."""
resp = admin_client.get('/org-sync/configs')
data = resp.json()['data']
assert isinstance(data, list)
ids = [c['id'] for c in data]
assert config_id in ids
def test_get_config_detail(self, admin_client, config_id):
"""AC-04: detail with masked auth_config."""
resp = admin_client.get(f'/org-sync/configs/{config_id}')
data = resp.json()['data']
assert data['auth_config']['api_key'] == '****'
def test_auth_config_masking(self, admin_client, config_id):
"""AC-34: sensitive fields masked in responses."""
resp = admin_client.get(f'/org-sync/configs/{config_id}')
auth = resp.json()['data']['auth_config']
assert auth.get('api_key') == '****'
# Non-sensitive fields should be visible
assert 'endpoint_url' in auth
def test_update_config(self, admin_client, config_id):
"""AC-05: update schedule_type."""
resp = admin_client.put(f'/org-sync/configs/{config_id}', json={
'schedule_type': 'cron',
'cron_expression': '0 3 * * *',
})
data = resp.json()['data']
assert data['schedule_type'] == 'cron'
assert data['cron_expression'] == '0 3 * * *'
# Restore
admin_client.put(f'/org-sync/configs/{config_id}', json={
'schedule_type': 'manual',
'cron_expression': None,
})
def test_update_merge_auth_config(self, admin_client, config_id):
"""AC-06: auth_config merge update."""
# Update only api_key
resp = admin_client.put(f'/org-sync/configs/{config_id}', json={
'auth_config': {'api_key': 'new-key-99999'},
})
data = resp.json()['data']
# api_key should be masked
assert data['auth_config']['api_key'] == '****'
# Other fields should persist
assert data['auth_config'].get('endpoint_url') is not None
def test_get_nonexistent_config(self, admin_client):
"""AC-08: nonexistent config → 22000."""
resp = admin_client.get('/org-sync/configs/999999')
data = resp.json()
assert data['status_code'] == 22000
# ---------------------------------------------------------------------------
# Exec endpoint tests (limited without real provider)
# ---------------------------------------------------------------------------
class TestExecEndpoints:
def test_get_logs_empty(self, admin_client, config_id):
"""AC-29: logs for a config with no syncs."""
resp = admin_client.get(f'/org-sync/configs/{config_id}/logs')
data = resp.json()['data']
assert 'data' in data
assert 'total' in data
def test_test_connection_wecom(self, admin_client):
"""AC-12: WeChat Work provider not implemented → 22004."""
# Create a wecom config
resp = admin_client.post('/org-sync/configs', json={
'provider': 'wecom',
'config_name': f'{PREFIX}wecom-stub',
'auth_type': 'api_key',
'auth_config': {'app_id': 'test', 'app_secret': 'test'},
})
wecom_id = resp.json()['data']['id']
try:
resp = admin_client.post(f'/org-sync/configs/{wecom_id}/test')
data = resp.json()
assert data['status_code'] == 22004
finally:
admin_client.delete(f'/org-sync/configs/{wecom_id}')
# ---------------------------------------------------------------------------
# Delete test
# ---------------------------------------------------------------------------
class TestDelete:
def test_delete_config(self, admin_client):
"""AC-07: soft delete."""
# Create a config to delete
resp = admin_client.post('/org-sync/configs', json={
'provider': 'feishu',
'config_name': f'{PREFIX}to-delete',
'auth_type': 'api_key',
'auth_config': {'app_id': 'x', 'app_secret': 'y'},
})
cid = resp.json()['data']['id']
resp = admin_client.delete(f'/org-sync/configs/{cid}')
assert resp.json()['status_code'] == 200
# Should no longer appear in list
resp = admin_client.get(f'/org-sync/configs/{cid}')
assert resp.json()['status_code'] == 22000
+148
View File
@@ -0,0 +1,148 @@
"""API integration tests for F009 org sync — test DTO validation and masking.
These tests verify request/response schemas and the masking function
without requiring a running server or database.
"""
import pytest
from bisheng.org_sync.domain.schemas.org_sync_schema import (
OrgSyncConfigCreate,
OrgSyncConfigUpdate,
OrgSyncConfigRead,
OrgSyncLogRead,
RemoteTreeNode,
mask_sensitive_fields,
)
class TestOrgSyncConfigCreate:
def test_valid_config(self):
config = OrgSyncConfigCreate(
provider='feishu',
config_name='Test',
auth_type='api_key',
auth_config={'app_id': 'xxx', 'app_secret': 'yyy'},
)
assert config.provider == 'feishu'
assert config.schedule_type == 'manual'
def test_invalid_provider(self):
with pytest.raises(ValueError, match='provider must be one of'):
OrgSyncConfigCreate(
provider='invalid',
config_name='Test',
auth_type='api_key',
auth_config={},
)
def test_invalid_auth_type(self):
with pytest.raises(ValueError, match='auth_type must be one of'):
OrgSyncConfigCreate(
provider='feishu',
config_name='Test',
auth_type='oauth',
auth_config={},
)
def test_cron_schedule(self):
config = OrgSyncConfigCreate(
provider='generic_api',
config_name='Test',
auth_type='api_key',
auth_config={'endpoint_url': 'https://api.example.com'},
schedule_type='cron',
cron_expression='0 2 * * *',
)
assert config.schedule_type == 'cron'
assert config.cron_expression == '0 2 * * *'
class TestOrgSyncConfigUpdate:
def test_partial_update(self):
update = OrgSyncConfigUpdate(schedule_type='manual')
assert update.schedule_type == 'manual'
assert update.auth_config is None
assert update.status is None
def test_full_update(self):
update = OrgSyncConfigUpdate(
auth_config={'app_secret': 'new'},
schedule_type='cron',
cron_expression='0 3 * * *',
status='disabled',
)
assert update.auth_config == {'app_secret': 'new'}
class TestMaskSensitiveFields:
def test_mask_known_keys(self):
auth = {
'app_id': 'cli_xxx',
'app_secret': 'secret123',
'api_key': 'sk-xxx',
'password': 'mypassword',
}
masked = mask_sensitive_fields(auth)
assert masked['app_id'] == 'cli_xxx'
assert masked['app_secret'] == '****'
assert masked['api_key'] == '****'
assert masked['password'] == '****'
def test_mask_nested(self):
auth = {
'outer': 'visible',
'nested': {'password': 'hidden', 'name': 'visible'},
}
masked = mask_sensitive_fields(auth)
assert masked['outer'] == 'visible'
assert masked['nested']['password'] == '****'
assert masked['nested']['name'] == 'visible'
def test_empty_dict(self):
assert mask_sensitive_fields({}) == {}
class TestOrgSyncConfigRead:
def test_serialization(self):
read = OrgSyncConfigRead(
id=1,
provider='feishu',
config_name='Test',
auth_type='api_key',
auth_config={'app_id': 'xxx', 'app_secret': '****'},
schedule_type='manual',
sync_status='idle',
status='active',
)
data = read.model_dump(mode='json')
assert data['id'] == 1
assert data['auth_config']['app_secret'] == '****'
class TestOrgSyncLogRead:
def test_defaults(self):
log = OrgSyncLogRead(
id=1, config_id=1, trigger_type='manual', status='running',
)
assert log.dept_created == 0
assert log.error_details is None
class TestRemoteTreeNode:
def test_tree_structure(self):
root = RemoteTreeNode(
external_id='root',
name='Root',
children=[
RemoteTreeNode(external_id='child', name='Child'),
],
)
assert len(root.children) == 1
assert root.children[0].external_id == 'child'
@@ -0,0 +1,258 @@
"""Unit tests for org sync Reconciler — pure logic, no IO.
Covers AC-16 through AC-28 (department + member reconciliation).
"""
import pytest
from unittest.mock import MagicMock
from bisheng.org_sync.domain.schemas.remote_dto import RemoteDepartmentDTO, RemoteMemberDTO
from bisheng.org_sync.domain.services.reconciler import (
ArchiveDept,
CreateDept,
CreateMember,
DisableMember,
MoveDept,
ReactivateMember,
TransferMember,
UpdateDept,
UpdateMember,
reconcile_departments,
reconcile_members,
)
# ---------------------------------------------------------------------------
# Helper: mock Department / User objects
# ---------------------------------------------------------------------------
def _make_dept(id, name, parent_id=None, source='feishu', external_id=None, path='/', status='active'):
dept = MagicMock()
dept.id = id
dept.name = name
dept.parent_id = parent_id
dept.source = source
dept.external_id = external_id
dept.path = path
dept.status = status
return dept
def _make_user(user_id, user_name, source='feishu', external_id=None, delete=0, email=None, phone_number=None):
user = MagicMock()
user.user_id = user_id
user.user_name = user_name
user.source = source
user.external_id = external_id
user.delete = delete
user.email = email
user.phone_number = phone_number
return user
def _make_user_dept(user_id, department_id, is_primary=1):
ud = MagicMock()
ud.user_id = user_id
ud.department_id = department_id
ud.is_primary = is_primary
return ud
# ===========================================================================
# Department Tests
# ===========================================================================
class TestReconcileDepartments:
def test_dept_create(self):
"""AC-16: remote has new department → CreateDept."""
remote = [RemoteDepartmentDTO(external_id='d1', name='Dev')]
local = []
ops = reconcile_departments(remote, local, 'feishu')
assert len(ops) == 1
assert isinstance(ops[0], CreateDept)
assert ops[0].remote.external_id == 'd1'
def test_dept_rename_third_party(self):
"""AC-17: third-party sourced dept renamed → UpdateDept."""
remote = [RemoteDepartmentDTO(external_id='d1', name='Engineering')]
local = [_make_dept(1, 'Dev', source='feishu', external_id='d1')]
ops = reconcile_departments(remote, local, 'feishu')
updates = [o for o in ops if isinstance(o, UpdateDept)]
assert len(updates) == 1
assert updates[0].new_name == 'Engineering'
assert updates[0].change_source is False
def test_dept_rename_local(self):
"""AC-18: local-sourced dept matched by external_id → force overwrite."""
remote = [RemoteDepartmentDTO(external_id='d1', name='Engineering')]
local = [_make_dept(1, 'Engineering', source='local', external_id='d1')]
ops = reconcile_departments(remote, local, 'feishu')
updates = [o for o in ops if isinstance(o, UpdateDept)]
assert len(updates) == 1
assert updates[0].change_source is True
def test_dept_move(self):
"""AC-19: department parent changed → MoveDept."""
remote = [
RemoteDepartmentDTO(external_id='d1', name='Dev', parent_external_id=None),
RemoteDepartmentDTO(external_id='d2', name='QA', parent_external_id='d1'),
]
# d2 is currently under d3 locally, but remote says under d1
local = [
_make_dept(1, 'Dev', parent_id=None, source='feishu', external_id='d1', path='/1/'),
_make_dept(2, 'QA', parent_id=3, source='feishu', external_id='d2', path='/3/2/'),
_make_dept(3, 'Old', parent_id=None, source='feishu', external_id='d3', path='/3/'),
]
ops = reconcile_departments(remote, local, 'feishu')
moves = [o for o in ops if isinstance(o, MoveDept)]
assert len(moves) == 1
assert moves[0].new_parent_external_id == 'd1'
def test_dept_archive(self):
"""AC-20: remote dept disappears → ArchiveDept."""
remote = []
local = [_make_dept(1, 'Dev', source='feishu', external_id='d1')]
ops = reconcile_departments(remote, local, 'feishu')
archives = [o for o in ops if isinstance(o, ArchiveDept)]
assert len(archives) == 1
assert archives[0].local.id == 1
def test_dept_archive_cascade(self):
"""AC-21: archived dept with local child → child also archived."""
remote = []
local = [
_make_dept(1, 'Dev', parent_id=None, source='feishu', external_id='d1', path='/1/'),
_make_dept(2, 'SubDev', parent_id=1, source='local', external_id=None, path='/1/2/'),
]
ops = reconcile_departments(remote, local, 'feishu')
archives = [o for o in ops if isinstance(o, ArchiveDept)]
assert len(archives) == 2
def test_dept_topological_order(self):
"""Create operations should be parent-first."""
remote = [
RemoteDepartmentDTO(external_id='child', name='Child', parent_external_id='parent'),
RemoteDepartmentDTO(external_id='parent', name='Parent'),
]
ops = reconcile_departments(remote, [], 'feishu')
creates = [o for o in ops if isinstance(o, CreateDept)]
assert len(creates) == 2
assert creates[0].remote.external_id == 'parent'
assert creates[1].remote.external_id == 'child'
def test_dept_cycle_detection(self):
"""Circular references in create deps → skip affected nodes."""
remote = [
RemoteDepartmentDTO(external_id='a', name='A', parent_external_id='b'),
RemoteDepartmentDTO(external_id='b', name='B', parent_external_id='a'),
]
ops = reconcile_departments(remote, [], 'feishu')
creates = [o for o in ops if isinstance(o, CreateDept)]
# Cycle: both have in-degree 1, neither starts at 0 → both skipped
assert len(creates) == 0
def test_dept_empty_input(self):
"""No remote, no local → no operations."""
ops = reconcile_departments([], [], 'feishu')
assert len(ops) == 0
# ===========================================================================
# Member Tests
# ===========================================================================
class TestReconcileMembers:
def test_member_create(self):
"""AC-22: new remote employee → CreateMember."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice', primary_dept_external_id='d1')]
ops = reconcile_members(remote, [], {}, {'d1': 1}, 'feishu')
assert len(ops) == 1
assert isinstance(ops[0], CreateMember)
def test_member_update(self):
"""AC-24: info changed → UpdateMember."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice New', email='new@x.com', primary_dept_external_id='d1')]
local = [_make_user(1, 'Alice', source='feishu', external_id='m1', email='old@x.com')]
user_depts = {1: [_make_user_dept(1, 10, is_primary=1)]}
ops = reconcile_members(remote, local, user_depts, {'d1': 10}, 'feishu')
updates = [o for o in ops if isinstance(o, UpdateMember)]
assert len(updates) == 1
assert updates[0].new_name == 'Alice New'
assert updates[0].new_email == 'new@x.com'
def test_member_transfer(self):
"""AC-25: primary department changed → TransferMember."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice', primary_dept_external_id='d2')]
local = [_make_user(1, 'Alice', source='feishu', external_id='m1')]
user_depts = {1: [_make_user_dept(1, 10, is_primary=1)]}
ext_map = {'d1': 10, 'd2': 20}
ops = reconcile_members(remote, local, user_depts, ext_map, 'feishu')
transfers = [o for o in ops if isinstance(o, TransferMember)]
assert len(transfers) == 1
assert transfers[0].new_primary_dept_external_id == 'd2'
assert transfers[0].old_primary_dept_id == 10
def test_member_secondary_dept_change(self):
"""AC-26: secondary department added/removed."""
remote = [RemoteMemberDTO(
external_id='m1', name='Alice',
primary_dept_external_id='d1',
secondary_dept_external_ids=['d3'], # add d3, remove d2
)]
local = [_make_user(1, 'Alice', source='feishu', external_id='m1')]
user_depts = {1: [
_make_user_dept(1, 10, is_primary=1),
_make_user_dept(1, 20, is_primary=0), # d2 secondary, to be removed
]}
ext_map = {'d1': 10, 'd2': 20, 'd3': 30}
ops = reconcile_members(remote, local, user_depts, ext_map, 'feishu')
transfers = [o for o in ops if isinstance(o, TransferMember)]
assert len(transfers) == 1
assert 30 not in transfers[0].remove_secondary_dept_ids
assert 20 in transfers[0].remove_secondary_dept_ids
def test_member_disable(self):
"""AC-27: employee disappeared from remote → DisableMember."""
remote = []
local = [_make_user(1, 'Alice', source='feishu', external_id='m1', delete=0)]
user_depts = {1: [_make_user_dept(1, 10, is_primary=1)]}
ops = reconcile_members(remote, local, user_depts, {'d1': 10}, 'feishu')
disables = [o for o in ops if isinstance(o, DisableMember)]
assert len(disables) == 1
assert disables[0].user_id == 1
def test_member_reactivate(self):
"""AC-28: previously disabled user reappears → ReactivateMember."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice', primary_dept_external_id='d1')]
local = [_make_user(1, 'Alice', source='feishu', external_id='m1', delete=1)]
user_depts = {1: []}
ops = reconcile_members(remote, local, user_depts, {'d1': 10}, 'feishu')
reactivates = [o for o in ops if isinstance(o, ReactivateMember)]
assert len(reactivates) == 1
assert reactivates[0].user_id == 1
def test_member_local_conflict(self):
"""Local user with matching external_id → force overwrite source."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice Updated', primary_dept_external_id='d1')]
local = [_make_user(1, 'Alice', source='local', external_id='m1')]
user_depts = {1: [_make_user_dept(1, 10, is_primary=1)]}
ops = reconcile_members(remote, local, user_depts, {'d1': 10}, 'feishu')
updates = [o for o in ops if isinstance(o, UpdateMember)]
assert len(updates) == 1
assert updates[0].change_source is True
def test_member_empty_input(self):
"""No remote, no local → no operations."""
ops = reconcile_members([], [], {}, {}, 'feishu')
assert len(ops) == 0
def test_member_disabled_remote_status(self):
"""Remote status=disabled → DisableMember even if local is active."""
remote = [RemoteMemberDTO(external_id='m1', name='Alice', status='disabled', primary_dept_external_id='d1')]
local = [_make_user(1, 'Alice', source='feishu', external_id='m1', delete=0)]
user_depts = {1: [_make_user_dept(1, 10, is_primary=1)]}
ops = reconcile_members(remote, local, user_depts, {'d1': 10}, 'feishu')
disables = [o for o in ops if isinstance(o, DisableMember)]
assert len(disables) == 1