ocrd_network.server_utils module

ocrd_network.server_utils.create_processing_message(logger: Logger, job: DBProcessorJob) → OcrdProcessingMessage

async ocrd_network.server_utils.create_workspace_if_not_exists(logger: Logger, mets_path: str) → DBWorkspace

async ocrd_network.server_utils.get_from_database_workflow_job(logger: Logger, workflow_job_id: str) → DBWorkflowJob

async ocrd_network.server_utils.get_from_database_workspace(logger: Logger, workspace_id: str | None = None, workspace_mets_path: str | None = None) → DBWorkspace

ocrd_network.server_utils.get_page_ids_list(logger: Logger, mets_path: str, page_id: str) → List[str]

ocrd_network.server_utils.request_processor_server_tool_json(logger: Logger, processor_server_base_url: str) → Dict

async ocrd_network.server_utils.forward_job_to_processor_server(logger: Logger, job_input: PYJobInput, processor_server_base_url: str) → PYJobOutput

async ocrd_network.server_utils.get_workflow_content(logger: Logger, workflow_id: str, workflow: UploadFile | None) → str

async ocrd_network.server_utils.validate_and_return_mets_path(logger: Logger, job_input: PYJobInput) → str

ocrd_network.server_utils.parse_workflow_tasks(logger: Logger, workflow_content: str) → List[ProcessorTask]

ocrd_network.server_utils.raise_http_exception(logger: Logger, status_code: int, message: str, error: Exception | None = None) → None

ocrd_network.server_utils.validate_job_input(logger: Logger, processor_name: str, ocrd_tool: dict, job_input: PYJobInput) → None

ocrd_network.server_utils.validate_workflow(logger: Logger, workflow: str) → None

Check whether workflow is not empty and parseable to a lists of ProcessorTask

ocrd_network.server_utils.validate_first_task_input_file_groups_existence(logger: Logger, mets_path: str, input_file_grps: List[str])