Skip to content

feat(TaskProcessingApi): Add endpoint for getting the next task #45391

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
provokateurin merged 2 commits into from
Jul 1, 2024
Merged

feat(TaskProcessingApi): Add endpoint for getting the next task #45391

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5aefdc3
feat(AppFramework): Add ExAppRequired attribute
provokateurin Jun 7, 2024
f5ff813
feat(TaskProcessingApi): Add endpoint for getting the next task
provokateurin May 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 137 additions & 46 deletions core/Controller/TaskProcessingApiController.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,33 +14,41 @@
use OCP\AppFramework\Http;
use OCP\AppFramework\Http\Attribute\AnonRateLimit;
use OCP\AppFramework\Http\Attribute\ApiRoute;
use OCP\AppFramework\Http\Attribute\ExAppRequired;
use OCP\AppFramework\Http\Attribute\NoAdminRequired;
use OCP\AppFramework\Http\Attribute\PublicPage;
use OCP\AppFramework\Http\Attribute\UserRateLimit;
use OCP\AppFramework\Http\DataDownloadResponse;
use OCP\AppFramework\Http\DataResponse;
use OCP\Files\File;
use OCP\Files\GenericFileException;
use OCP\Files\IRootFolder;
use OCP\Files\NotPermittedException;
use OCP\IL10N;
use OCP\IRequest;
use OCP\Lock\LockedException;
use OCP\TaskProcessing\EShapeType;
use OCP\TaskProcessing\Exception\Exception;
use OCP\TaskProcessing\Exception\NotFoundException;
use OCP\TaskProcessing\Exception\PreConditionNotMetException;
use OCP\TaskProcessing\Exception\UnauthorizedException;
use OCP\TaskProcessing\Exception\ValidationException;
use OCP\TaskProcessing\IManager;
use OCP\TaskProcessing\ShapeDescriptor;
use OCP\TaskProcessing\Task;
use RuntimeException;

/**
* @psalm-import-type CoreTaskProcessingTask from ResponseDefinitions
* @psalm-import-type CoreTaskProcessingTaskType from ResponseDefinitions
*/
class TaskProcessingApiController extends \OCP\AppFramework\OCSController {
public function __construct(
string $appName,
IRequest $request,
private \OCP\TaskProcessing\IManager $taskProcessingManager,
private IL10N $l,
private ?string $userId,
string $appName,
IRequest $request,
private IManager $taskProcessingManager,
private IL10N $l,
private ?string $userId,
private IRootFolder $rootFolder,
) {
parent::__construct($appName, $request);
Expand Down Expand Up @@ -109,13 +117,13 @@ public function schedule(array $input, string $type, string $appId, string $cust
return new DataResponse([
'task' => $json,
]);
} catch (\OCP\TaskProcessing\Exception\PreConditionNotMetException) {
} catch (PreConditionNotMetException) {
return new DataResponse(['message' => $this->l->t('The given provider is not available')], Http::STATUS_PRECONDITION_FAILED);
} catch (ValidationException $e) {
return new DataResponse(['message' => $e->getMessage()], Http::STATUS_BAD_REQUEST);
} catch (UnauthorizedException $e) {
} catch (UnauthorizedException) {
return new DataResponse(['message' => 'User does not have access to the files mentioned in the task input'], Http::STATUS_UNAUTHORIZED);
} catch (\OCP\TaskProcessing\Exception\Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => 'Internal server error'], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand Down Expand Up @@ -144,9 +152,9 @@ public function getTask(int $id): DataResponse {
return new DataResponse([
'task' => $json,
]);
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Task not found')], Http::STATUS_NOT_FOUND);
} catch (\RuntimeException $e) {
} catch (RuntimeException) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand All @@ -169,9 +177,9 @@ public function deleteTask(int $id): DataResponse {
$this->taskProcessingManager->deleteTask($task);

return new DataResponse(null);
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
} catch (NotFoundException) {
return new DataResponse(null);
} catch (\OCP\TaskProcessing\Exception\Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand Down Expand Up @@ -199,7 +207,7 @@ public function listTasksByApp(string $appId, ?string $customId = null): DataRes
return new DataResponse([
'tasks' => $json,
]);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand All @@ -226,7 +234,7 @@ public function listTasks(?string $taskType, ?string $customId = null): DataResp
return new DataResponse([
'tasks' => $json,
]);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand All @@ -247,37 +255,72 @@ public function listTasks(?string $taskType, ?string $customId = null): DataResp
public function getFileContents(int $taskId, int $fileId): Http\DataDownloadResponse|DataResponse {
try {
$task = $this->taskProcessingManager->getUserTask($taskId, $this->userId);
$ids = $this->extractFileIdsFromTask($task);
if (!in_array($fileId, $ids)) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
}
$node = $this->rootFolder->getFirstNodeById($fileId);
if ($node === null) {
$node = $this->rootFolder->getFirstNodeByIdInPath($fileId, '/' . $this->rootFolder->getAppDataDirectoryName() . '/');
if (!$node instanceof File) {
throw new \OCP\TaskProcessing\Exception\NotFoundException('Node is not a file');
}
} elseif (!$node instanceof File) {
throw new \OCP\TaskProcessing\Exception\NotFoundException('Node is not a file');
}
return new Http\DataDownloadResponse($node->getContent(), $node->getName(), $node->getMimeType());
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
return $this->getFileContentsInternal($task, $fileId);
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}

/**
* Returns the contents of a file referenced in a task(ExApp route version)
*
* @param int $taskId The id of the task
* @param int $fileId The file id of the file to retrieve
* @return DataDownloadResponse<Http::STATUS_OK, string, array{}>|DataResponse<Http::STATUS_INTERNAL_SERVER_ERROR|Http::STATUS_NOT_FOUND, array{message: string}, array{}>
*
* 200: File content returned
* 404: Task or file not found
*/
#[ExAppRequired]
#[ApiRoute(verb: 'GET', url: '/tasks_provider/{taskId}/file/{fileId}', root: '/taskprocessing')]
public function getFileContentsExApp(int $taskId, int $fileId): Http\DataDownloadResponse|DataResponse {
try {
$task = $this->taskProcessingManager->getTask($taskId);
return $this->getFileContentsInternal($task, $fileId);
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}

/**
* @throws NotPermittedException
* @throws NotFoundException
* @throws GenericFileException
* @throws LockedException
*
* @return DataDownloadResponse<Http::STATUS_OK, string, array{}>|DataResponse<Http::STATUS_INTERNAL_SERVER_ERROR|Http::STATUS_NOT_FOUND, array{message: string}, array{}>
*/
private function getFileContentsInternal(Task $task, int $fileId): Http\DataDownloadResponse|DataResponse {
$ids = $this->extractFileIdsFromTask($task);
if (!in_array($fileId, $ids)) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
}
$node = $this->rootFolder->getFirstNodeById($fileId);
if ($node === null) {
$node = $this->rootFolder->getFirstNodeByIdInPath($fileId, '/' . $this->rootFolder->getAppDataDirectoryName() . '/');
if (!$node instanceof File) {
throw new NotFoundException('Node is not a file');
}
} elseif (!$node instanceof File) {
throw new NotFoundException('Node is not a file');
}
return new Http\DataDownloadResponse($node->getContent(), $node->getName(), $node->getMimeType());
}

/**
* @param Task $task
* @return list<int>
* @throws \OCP\TaskProcessing\Exception\NotFoundException
* @throws NotFoundException
*/
private function extractFileIdsFromTask(Task $task): array {
$ids = [];
$taskTypes = $this->taskProcessingManager->getAvailableTaskTypes();
if (!isset($taskTypes[$task->getTaskTypeId()])) {
throw new \OCP\TaskProcessing\Exception\NotFoundException('Could not find task type');
throw new NotFoundException('Could not find task type');
}
$taskType = $taskTypes[$task->getTaskTypeId()];
foreach ($taskType['inputShape'] + $taskType['optionalInputShape'] as $key => $descriptor) {
Expand Down Expand Up @@ -317,22 +360,22 @@ private function extractFileIdsFromTask(Task $task): array {
* 200: Progress updated successfully
* 404: Task not found
*/
#[NoAdminRequired]
#[ApiRoute(verb: 'POST', url: '/tasks/{taskId}/progress', root: '/taskprocessing')]
#[ExAppRequired]
#[ApiRoute(verb: 'POST', url: '/tasks_provider/{taskId}/progress', root: '/taskprocessing')]
public function setProgress(int $taskId, float $progress): DataResponse {
try {
$this->taskProcessingManager->setTaskProgress($taskId, $progress);
$task = $this->taskProcessingManager->getUserTask($taskId, $this->userId);
$task = $this->taskProcessingManager->getTask($taskId);

/** @var CoreTaskProcessingTask $json */
$json = $task->jsonSerialize();

return new DataResponse([
'task' => $json,
]);
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand All @@ -348,25 +391,23 @@ public function setProgress(int $taskId, float $progress): DataResponse {
* 200: Result updated successfully
* 404: Task not found
*/
#[NoAdminRequired]
#[ApiRoute(verb: 'POST', url: '/tasks/{taskId}/result', root: '/taskprocessing')]
#[ExAppRequired]
#[ApiRoute(verb: 'POST', url: '/tasks_provider/{taskId}/result', root: '/taskprocessing')]
public function setResult(int $taskId, ?array $output = null, ?string $errorMessage = null): DataResponse {
try {
// Check if the current user can access the task
$this->taskProcessingManager->getUserTask($taskId, $this->userId);
// set result
$this->taskProcessingManager->setTaskResult($taskId, $errorMessage, $output);
$task = $this->taskProcessingManager->getUserTask($taskId, $this->userId);
$task = $this->taskProcessingManager->getTask($taskId);

/** @var CoreTaskProcessingTask $json */
$json = $task->jsonSerialize();

return new DataResponse([
'task' => $json,
]);
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand Down Expand Up @@ -396,9 +437,59 @@ public function cancelTask(int $taskId): DataResponse {
return new DataResponse([
'task' => $json,
]);
} catch (\OCP\TaskProcessing\Exception\NotFoundException $e) {
} catch (NotFoundException) {
return new DataResponse(['message' => $this->l->t('Not found')], Http::STATUS_NOT_FOUND);
} catch (Exception $e) {
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}

/**
* Returns the next scheduled task for the taskTypeId
*
* @param list<string> $providerIds The ids of the providers
* @param list<string> $taskTypeIds The ids of the task types
* @return DataResponse<Http::STATUS_OK, array{task: CoreTaskProcessingTask, provider: array{name: string}}, array{}>|DataResponse<Http::STATUS_NO_CONTENT, null, array{}>|DataResponse<Http::STATUS_INTERNAL_SERVER_ERROR, array{message: string}, array{}>
*
* 200: Task returned
* 204: No task found
*/
#[ExAppRequired]
#[ApiRoute(verb: 'GET', url: '/tasks_provider/next', root: '/taskprocessing')]
public function getNextScheduledTask(array $providerIds, array $taskTypeIds): DataResponse {
try {
// restrict $providerIds to providers that are configured as preferred for the passed task types
$providerIds = array_values(array_intersect(array_unique(array_map(fn ($taskTypeId) => $this->taskProcessingManager->getPreferredProvider($taskTypeId)->getId(), $taskTypeIds)), $providerIds));
// restrict $taskTypeIds to task types that can actually be run by one of the now restricted providers
$taskTypeIds = array_values(array_filter($taskTypeIds, fn ($taskTypeId) => in_array($this->taskProcessingManager->getPreferredProvider($taskTypeId)->getId(), $providerIds, true)));
if (count($providerIds) === 0 || count($taskTypeIds) === 0) {
throw new NotFoundException();
}

$taskIdsToIgnore = [];
while (true) {
$task = $this->taskProcessingManager->getNextScheduledTask($taskTypeIds, $taskIdsToIgnore);
$provider = $this->taskProcessingManager->getPreferredProvider($task->getTaskTypeId());
if (in_array($provider->getId(), $providerIds, true)) {
if ($this->taskProcessingManager->lockTask($task)) {
break;
}
}
$taskIdsToIgnore[] = (int)$task->getId();
}

/** @var CoreTaskProcessingTask $json */
$json = $task->jsonSerialize();

return new DataResponse([
'task' => $json,
'provider' => [
'name' => $provider->getId(),
],
]);
} catch (NotFoundException) {
return new DataResponse(null, Http::STATUS_NO_CONTENT);
} catch (Exception) {
return new DataResponse(['message' => $this->l->t('Internal error')], Http::STATUS_INTERNAL_SERVER_ERROR);
}
}
Expand Down
Loading