chore: implement prefect task scheduler

[iziang]

No description provided.

[cursor]

Bug: Prefect Scheduler Incorrectly Retrieves Flow Run ID

The PrefectTaskScheduler methods (e.g., schedule_create_index, schedule_collection_init, schedule_collection_delete, schedule_reconcile_indexes) incorrectly attempt to retrieve a flow run ID by accessing a .id attribute on the return value of Prefect flow functions. These flow functions are designed to return dictionaries, not objects with an .id attribute. While a hasattr check prevents an AttributeError, it causes the flow_run_id to be set to the string representation of the returned dictionary, which is not a valid or trackable Prefect flow run ID.

aperag/tasks/scheduler.py#L434-L440

ApeRAG/aperag/tasks/scheduler.py

Lines 434 to 440 in c9de47d

	# Execute workflow and return flow run ID
	flow_state = create_document_indexes_flow(document_id, index_types)
	flow_run_id = flow_state.id if hasattr(flow_state, 'id') else str(flow_state)
	logger.debug(
	f"Scheduled create indexes flow {flow_run_id} for document {document_id} with types {index_types}"
	)
	return flow_run_id

aperag/tasks/scheduler.py#L466-L471

ApeRAG/aperag/tasks/scheduler.py

Lines 466 to 471 in c9de47d

	# Execute workflow and return flow run ID
	flow_state = delete_document_indexes_flow(document_id, index_types)
	flow_run_id = flow_state.id if hasattr(flow_state, 'id') else str(flow_state)
	logger.debug(
	f"Scheduled delete indexes flow {flow_run_id} for document {document_id} with types {index_types}"
	)

aperag/tasks/scheduler.py#L482-L484

ApeRAG/aperag/tasks/scheduler.py

Lines 482 to 484 in c9de47d

	flow_state = collection_init_flow(collection_id, document_user_quota)
	flow_run_id = flow_state.id if hasattr(flow_state, 'id') else str(flow_state)
	logger.debug(f"Scheduled collection init flow {flow_run_id} for collection {collection_id}")

Fix in Cursor

---

Bug: Quota Initialization Bug

The if not document_user_quota: condition incorrectly defaults document_user_quota to 100. This changes valid quota values of 0 (meaning no documents allowed) and None (potentially signifying an unlimited quota) to 100, fundamentally altering the intended collection initialization behavior.

aperag/service/collection_service.py#L91-L94

ApeRAG/aperag/service/collection_service.py

Lines 91 to 94 in c9de47d

	document_user_quota = await self.db_ops.query_user_quota(user, QuotaType.MAX_DOCUMENT_COUNT)
	if not document_user_quota:
	document_user_quota = 100

Fix in Cursor

---

Bug: Flow State Access Error in Scheduler Methods

Incorrect attribute access pattern in PrefectTaskScheduler methods, including schedule_update_index and schedule_collection_delete. The code attempts to access flow_state.id when flow_state can be a dictionary, leading to an AttributeError.

aperag/tasks/scheduler.py#L450-L455

ApeRAG/aperag/tasks/scheduler.py

Lines 450 to 455 in c9de47d

	# Execute workflow and return flow run ID
	flow_state = update_document_indexes_flow(document_id, index_types)
	flow_run_id = flow_state.id if hasattr(flow_state, 'id') else str(flow_state)
	logger.debug(
	f"Scheduled update indexes flow {flow_run_id} for document {document_id} with types {index_types}"
	)

aperag/tasks/scheduler.py#L494-L496

ApeRAG/aperag/tasks/scheduler.py

Lines 494 to 496 in c9de47d

	try:
	flow_state = collection_delete_flow(collection_id)
	flow_run_id = flow_state.id if hasattr(flow_state, 'id') else str(flow_state)

Fix in Cursor

---

Was this report helpful? Give feedback by reacting with 👍 or 👎