Passed
Pull Request — master (#1083)
by
unknown
02:36
created

ocrd_network.database   B

Complexity

Total Complexity 46

Size/Duplication

Total Lines 178
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 46
eloc 129
dl 0
loc 178
rs 8.72
c 0
b 0
f 0

11 Functions

Rating   Name   Duplication   Size   Complexity  
A initiate_database() 0 5 1
A sync_db_update_processing_job() 0 3 1
A sync_initiate_database() 0 3 1
A db_create_workspace() 0 17 3
F db_update_workspace() 0 45 19
C db_update_processing_job() 0 24 9
B db_get_workspace() 0 17 7
A db_get_processing_job() 0 6 2
A sync_db_get_workspace() 0 3 1
A sync_db_get_processing_job() 0 3 1
A sync_db_update_workspace() 0 3 1

How to fix   Complexity   

Complexity

Complex classes like ocrd_network.database often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
""" The database is used to store information regarding jobs and workspaces.
2
3
Jobs: for every process-request a job is inserted into the database with an uuid, status and
4
information about the process like parameters and file groups. It is mainly used to track the status
5
(`ocrd_network.models.job.StateEnum`) of a job so that the state of a job can be queried. Finished
6
jobs are not deleted from the database.
7
8
Workspaces: A job or a processor always runs on a workspace. So a processor needs the information
9
where the workspace is available. This information can be set with providing an absolute path or a
10
workspace_id. With the latter, the database is used to convert the workspace_id to a path.
11
12
XXX: Currently the information is not preserved after the processing-server shuts down as the
13
database (runs in docker) currently has no volume set.
14
"""
15
from beanie import init_beanie
16
from motor.motor_asyncio import AsyncIOMotorClient
17
from uuid import uuid4
18
from pathlib import Path
19
20
from .models import (
21
    DBProcessorJob,
22
    DBWorkflowJob,
23
    DBWorkspace
24
)
25
from .utils import call_sync
26
27
28
async def initiate_database(db_url: str):
29
    client = AsyncIOMotorClient(db_url)
30
    await init_beanie(
31
        database=client.get_default_database(default='ocrd'),
32
        document_models=[DBProcessorJob, DBWorkflowJob, DBWorkspace]
33
    )
34
35
36
@call_sync
37
async def sync_initiate_database(db_url: str):
38
    await initiate_database(db_url)
39
40
41
async def db_create_workspace(mets_path: str) -> DBWorkspace:
42
    """ Create a workspace-database entry only from a mets-path
43
    """
44
    if not Path(mets_path).exists():
45
        raise FileNotFoundError(f'Cannot create DB workspace entry, `{mets_path}` does not exist!')
46
    try:
47
        return await db_get_workspace(workspace_mets_path=mets_path)
48
    except ValueError:
49
        workspace_db = DBWorkspace(
50
            workspace_id=str(uuid4()),
51
            workspace_path=Path(mets_path).parent,
52
            workspace_mets_path=mets_path,
53
            ocrd_identifier="",
54
            bagit_profile_identifier="",
55
        )
56
        await workspace_db.save()
57
        return workspace_db
58
59
60
async def db_get_workspace(workspace_id: str = None, workspace_mets_path: str = None) -> DBWorkspace:
61
    workspace = None
62
    if not workspace_id and not workspace_mets_path:
63
        raise ValueError(f'Either `workspace_id` or `workspace_mets_path` field must be used as a search key')
64
    if workspace_id:
65
        workspace = await DBWorkspace.find_one(
66
            DBWorkspace.workspace_id == workspace_id
67
        )
68
        if not workspace:
69
            raise ValueError(f'Workspace with id "{workspace_id}" not in the DB.')
70
    if workspace_mets_path:
71
        workspace = await DBWorkspace.find_one(
72
            DBWorkspace.workspace_mets_path == workspace_mets_path
73
        )
74
        if not workspace:
75
            raise ValueError(f'Workspace with path "{workspace_mets_path}" not in the DB.')
76
    return workspace
77
78
79
@call_sync
80
async def sync_db_get_workspace(workspace_id: str = None, workspace_mets_path: str = None) -> DBWorkspace:
81
    return await db_get_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path)
82
83
84
async def db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs) -> DBWorkspace:
85
    workspace = None
86
    if not workspace_id and not workspace_mets_path:
87
        raise ValueError(f'Either `workspace_id` or `workspace_mets_path` field must be used as a search key')
88
    if workspace_id:
89
        workspace = await DBWorkspace.find_one(
90
            DBWorkspace.workspace_id == workspace_id
91
        )
92
        if not workspace:
93
            raise ValueError(f'Workspace with id "{workspace_id}" not in the DB.')
94
    if workspace_mets_path:
95
        workspace = await DBWorkspace.find_one(
96
            DBWorkspace.workspace_mets_path == workspace_mets_path
97
        )
98
        if not workspace:
99
            raise ValueError(f'Workspace with path "{workspace_mets_path}" not in the DB.')
100
101
    job_keys = list(workspace.__dict__.keys())
102
    for key, value in kwargs.items():
103
        if key not in job_keys:
104
            raise ValueError(f'Field "{key}" is not available.')
105
        if key == 'workspace_id':
106
            workspace.workspace_id = value
107
        elif key == 'workspace_mets_path':
108
            workspace.workspace_mets_path = value
109
        elif key == 'ocrd_identifier':
110
            workspace.ocrd_identifier = value
111
        elif key == 'bagit_profile_identifier':
112
            workspace.bagit_profile_identifier = value
113
        elif key == 'ocrd_base_version_checksum':
114
            workspace.ocrd_base_version_checksum = value
115
        elif key == 'ocrd_mets':
116
            workspace.ocrd_mets = value
117
        elif key == 'bag_info_adds':
118
            workspace.bag_info_adds = value
119
        elif key == 'deleted':
120
            workspace.deleted = value
121
        elif key == 'pages_locked':
122
            workspace.pages_locked = value
123
        elif key == 'mets_server_url':
124
            workspace.mets_server_url = value
125
        else:
126
            raise ValueError(f'Field "{key}" is not updatable.')
127
    await workspace.save()
128
    return workspace
129
130
131
@call_sync
132
async def sync_db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs) -> DBWorkspace:
133
    return await db_update_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path, **kwargs)
134
135
136
async def db_get_processing_job(job_id: str) -> DBProcessorJob:
137
    job = await DBProcessorJob.find_one(
138
        DBProcessorJob.job_id == job_id)
139
    if not job:
140
        raise ValueError(f'Processing job with id "{job_id}" not in the DB.')
141
    return job
142
143
144
@call_sync
145
async def sync_db_get_processing_job(job_id: str) -> DBProcessorJob:
146
    return await db_get_processing_job(job_id)
147
148
149
async def db_update_processing_job(job_id: str, **kwargs) -> DBProcessorJob:
150
    job = await DBProcessorJob.find_one(
151
        DBProcessorJob.job_id == job_id)
152
    if not job:
153
        raise ValueError(f'Processing job with id "{job_id}" not in the DB.')
154
155
    job_keys = list(job.__dict__.keys())
156
    for key, value in kwargs.items():
157
        if key not in job_keys:
158
            raise ValueError(f'Field "{key}" is not available.')
159
        if key == 'state':
160
            job.state = value
161
        elif key == 'start_time':
162
            job.start_time = value
163
        elif key == 'end_time':
164
            job.end_time = value
165
        elif key == 'path_to_mets':
166
            job.path_to_mets = value
167
        elif key == 'exec_time':
168
            job.exec_time = value
169
        else:
170
            raise ValueError(f'Field "{key}" is not updatable.')
171
    await job.save()
172
    return job
173
174
175
@call_sync
176
async def sync_db_update_processing_job(job_id: str, **kwargs) -> DBProcessorJob:
177
    return await db_update_processing_job(job_id=job_id, **kwargs)
178