Merge pull request #1069 from OCR-D/processing_server_ext_1046

Processing server extension (#1046)

ocrd_network/ocrd_network/database.py

@@ -1,6 +1,6 @@
 """ The database is used to store information regarding jobs and workspaces.
 
-Jobs: for every process-request a job is inserted into the database with a uuid, status and
+Jobs: for every process-request a job is inserted into the database with an uuid, status and
 information about the process like parameters and file groups. It is mainly used to track the status
 (`ocrd_network.models.job.StateEnum`) of a job so that the state of a job can be queried. Finished
 jobs are not deleted from the database.
@@ -35,18 +35,77 @@ async def sync_initiate_database(db_url: str):
     await initiate_database(db_url)
 
 
-async def db_get_workspace(workspace_id: str) -> DBWorkspace:
-    workspace = await DBWorkspace.find_one(
-        DBWorkspace.workspace_id == workspace_id
-    )
-    if not workspace:
-        raise ValueError(f'Workspace with id "{workspace_id}" not in the DB.')
+async def db_get_workspace(workspace_id: str = None, workspace_mets_path: str = None) -> DBWorkspace:
+    workspace = None
+    if not workspace_id and not workspace_mets_path:
+        raise ValueError(f'Either `workspace_id` or `workspace_mets_path` field must be used as a search key')
+    if workspace_id:
+        workspace = await DBWorkspace.find_one(
+            DBWorkspace.workspace_id == workspace_id
+        )
+        if not workspace:
+            raise ValueError(f'Workspace with id "{workspace_id}" not in the DB.')
+    if workspace_mets_path:
+        workspace = await DBWorkspace.find_one(
+            DBWorkspace.workspace_mets_path == workspace_mets_path
+        )
+        if not workspace:
+            raise ValueError(f'Workspace with path "{workspace_mets_path}" not in the DB.')
     return workspace
 
 
 @call_sync
-async def sync_db_get_workspace(workspace_id: str) -> DBWorkspace:
-    return await db_get_workspace(workspace_id)
+async def sync_db_get_workspace(workspace_id: str = None, workspace_mets_path: str = None) -> DBWorkspace:
+    return await db_get_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path)
+
+
+async def db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs):
+    workspace = None
+    if not workspace_id and not workspace_mets_path:
+        raise ValueError(f'Either `workspace_id` or `workspace_mets_path` field must be used as a search key')
+    if workspace_id:
+        workspace = await DBWorkspace.find_one(
+            DBWorkspace.workspace_id == workspace_id
+        )
+        if not workspace:
+            raise ValueError(f'Workspace with id "{workspace_id}" not in the DB.')
+    if workspace_mets_path:
+        workspace = await DBWorkspace.find_one(
+            DBWorkspace.workspace_mets_path == workspace_mets_path
+        )
+        if not workspace:
+            raise ValueError(f'Workspace with path "{workspace_mets_path}" not in the DB.')
+
+    job_keys = list(workspace.__dict__.keys())
+    for key, value in kwargs.items():
+        if key not in job_keys:
+            raise ValueError(f'Field "{key}" is not available.')
+        if key == 'workspace_id':
+            workspace.workspace_id = value
+        elif key == 'workspace_mets_path':
+            workspace.workspace_mets_path = value
+        elif key == 'ocrd_identifier':
+            workspace.ocrd_identifier = value
+        elif key == 'bagit_profile_identifier':
+            workspace.bagit_profile_identifier = value
+        elif key == 'ocrd_base_version_checksum':
+            workspace.ocrd_base_version_checksum = value
+        elif key == 'ocrd_mets':
+            workspace.ocrd_mets = value
+        elif key == 'bag_info_adds':
+            workspace.bag_info_adds = value
+        elif key == 'deleted':
+            workspace.deleted = value
+        elif key == 'pages_locked':
+            workspace.pages_locked = value
+        else:
+            raise ValueError(f'Field "{key}" is not updatable.')
+    await workspace.save()
+
+
+@call_sync
+async def sync_db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs):
+    await db_update_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path, **kwargs)
 
 
 async def db_get_processing_job(job_id: str) -> DBProcessorJob:
@@ -68,8 +127,6 @@ async def db_update_processing_job(job_id: str, **kwargs):
     if not job:
         raise ValueError(f'Processing job with id "{job_id}" not in the DB.')
 
-    # TODO: This may not be the best Pythonic way to do it. However, it works!
-    #  There must be a shorter way with Pydantic. Suggest an improvement.
     job_keys = list(job.__dict__.keys())
     for key, value in kwargs.items():
         if key not in job_keys:

ocrd_network/ocrd_network/deployer.py

@@ -39,6 +39,7 @@ def __init__(self, config_path: str) -> None:
         self.data_mongo: DataMongoDB = DataMongoDB(config['database'])
         self.data_queue: DataRabbitMQ = DataRabbitMQ(config['process_queue'])
         self.data_hosts: List[DataHost] = []
+        self.internal_callback_url = config.get('internal_callback_url', None)
         for config_host in config['hosts']:
             self.data_hosts.append(DataHost(config_host))
 
@@ -302,7 +303,7 @@ def deploy_mongodb(
     ) -> str:
         if self.data_mongo.skip_deployment:
             self.log.debug('MongoDB is externaly managed. Skipping deployment')
-            verify_mongodb_available(self.data_mongo.url);
+            verify_mongodb_available(self.data_mongo.url)
             return self.data_mongo.url
 
         self.log.debug(f"Trying to deploy '{image}', with modes: "

ocrd_network/ocrd_network/models/__init__.py

@@ -9,6 +9,7 @@
     'PYJobInput',
     'PYJobOutput',
     'PYOcrdTool',
+    'PYResultMessage',
     'StateEnum',
 ]
 
@@ -18,5 +19,6 @@
     PYJobOutput,
     StateEnum
 )
+from .messages import PYResultMessage
 from .ocrd_tool import PYOcrdTool
 from .workspace import DBWorkspace

ocrd_network/ocrd_network/models/job.py

@@ -7,15 +7,22 @@
 
 
 class StateEnum(str, Enum):
+    # The processing job is cached inside the Processing Server requests cache
+    cached = 'CACHED'
+    # The processing job is queued inside the RabbitMQ
     queued = 'QUEUED'
+    # Processing job is currently running in a Worker or Processor Server
     running = 'RUNNING'
+    # Processing job finished successfully
     success = 'SUCCESS'
+    # Processing job failed
     failed = 'FAILED'
 
 
 class PYJobInput(BaseModel):
     """ Wraps the parameters required to make a run-processor-request
     """
+    processor_name: Optional[str] = None
     path_to_mets: Optional[str] = None
     workspace_id: Optional[str] = None
     description: Optional[str] = None
@@ -28,6 +35,10 @@ class PYJobInput(BaseModel):
     # Used to toggle between sending requests to 'worker and 'server',
     # i.e., Processing Worker and Processor Server, respectively
     agent_type: Optional[str] = 'worker'
+    # Auto generated by the Processing Server when forwarding to the Processor Server
+    job_id: Optional[str] = None
+    # If set, specifies a list of job ids this job depends on
+    depends_on: Optional[List[str]] = None
 
     class Config:
         schema_extra = {
@@ -65,8 +76,10 @@ class DBProcessorJob(Document):
     output_file_grps: Optional[List[str]]
     page_id: Optional[str]
     parameters: Optional[dict]
+    depends_on: Optional[List[str]]
     result_queue_name: Optional[str]
     callback_url: Optional[str]
+    internal_callback_url: Optional[str]
     start_time: Optional[datetime]
     end_time: Optional[datetime]
     exec_time: Optional[str]

ocrd_network/ocrd_network/models/messages.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+from typing import Optional
+from .job import StateEnum
+
+
+class PYResultMessage(BaseModel):
+    """ Wraps the parameters required to make a result message request
+    """
+    job_id: str
+    state: StateEnum
+    path_to_mets: Optional[str] = None
+    workspace_id: Optional[str] = None
+
+    class Config:
+        schema_extra = {
+            'example': {
+                'job_id': '123123123',
+                'state': 'SUCCESS',
+                'path_to_mets': '/path/to/mets.xml',
+                'workspace_id': 'c7f25615-fc17-4365-a74d-ad20e1ddbd0e'
+            }
+        }

ocrd_network/ocrd_network/models/workspace.py

@@ -1,5 +1,5 @@
 from beanie import Document
-from typing import Optional
+from typing import Dict, Optional
 
 
 class DBWorkspace(Document):
@@ -15,6 +15,10 @@ class DBWorkspace(Document):
         ocrd_mets                   Ocrd-Mets (optional)
         bag_info_adds               bag-info.txt can also (optionally) contain additional
                                     key-value-pairs which are saved here
+        deleted                     the document is deleted if set, however, the record is still preserved
+        pages_locked                a data structure that holds output `fileGrp`s and their respective locked `page_id`
+                                    that are currently being processed by an OCR-D processor (server or worker).
+                                    If no `page_id` field is set, an identifier "all_pages" will be used.
     """
     workspace_id: str
     workspace_mets_path: str
@@ -24,6 +28,10 @@ class DBWorkspace(Document):
     ocrd_mets: Optional[str]
     bag_info_adds: Optional[dict]
     deleted: bool = False
+    # Dictionary structure:
+    # Key: fileGrp
+    # Value: Set of `page_id`s
+    pages_locked: Optional[Dict] = {}
 
     class Settings:
         name = "workspace"