|
for i, (db_id, action_item) in enumerate(q_items.actions.items()): |
|
try: |
|
match action_item.type: |
|
case ActionType.DELETE_SOURCE_IDS: |
|
exec_in_proc(target=delete_by_source, args=(vectordb_loader, action_item.payload.sourceIds)) |
|
|
|
case ActionType.DELETE_PROVIDER_ID: |
|
exec_in_proc(target=delete_by_provider, args=(vectordb_loader, action_item.payload.providerId)) |
|
|
|
case ActionType.DELETE_USER_ID: |
|
exec_in_proc(target=delete_user, args=(vectordb_loader, action_item.payload.userId)) |
|
|
|
case ActionType.UPDATE_ACCESS_SOURCE_ID: |
|
exec_in_proc( |
|
target=update_access, |
|
args=( |
|
vectordb_loader, |
|
action_item.payload.op, |
|
action_item.payload.userIds, |
|
action_item.payload.sourceId, |
|
), |
|
) |
|
|
|
case ActionType.UPDATE_ACCESS_PROVIDER_ID: |
|
exec_in_proc( |
|
target=update_access_provider, |
|
args=( |
|
vectordb_loader, |
|
action_item.payload.op, |
|
action_item.payload.userIds, |
|
action_item.payload.providerId, |
|
), |
|
) |
|
|
|
case ActionType.UPDATE_ACCESS_DECL_SOURCE_ID: |
|
exec_in_proc( |
|
target=decl_update_access, |
|
args=( |
|
vectordb_loader, |
|
action_item.payload.userIds, |
|
action_item.payload.sourceId, |
|
), |
|
) |
|
|
|
case _: |
|
LOGGER.warning( |
|
f'Unknown action type {action_item.type} for action id {db_id},' |
|
f' type {action_item.type}, skipping and marking as processed', |
|
extra={ 'action_item': action_item }, |
|
) |
|
continue |
|
|
|
processed_event_ids.append(db_id) |
Instead of processing one type of update in a subprocess (which is from forkserver so is fast but is still a subprocess), bucket the list and send the batches in the service functions so all of the updates are done with the overhead of one subprocess
context_chat_backend/context_chat_backend/task_fetcher.py
Lines 346 to 398 in 9c525e5
https://github.com/nextcloud/context_chat_backend/blob/master/context_chat_backend/vectordb/service.py
Breathing time between updates/writes to the DB may still be required.