Passed
Pull Request — master (#1111)
by
unknown
02:28
created

ocrd_network.logging   A

Complexity

Total Complexity 9

Size/Duplication

Total Lines 48
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
wmc 9
eloc 29
dl 0
loc 48
rs 10
c 0
b 0
f 0

8 Functions

Rating   Name   Duplication   Size   Complexity  
A get_cache_processing_requests_logging_file_path() 0 2 1
A get_processor_server_logging_file_path() 0 2 1
A get_root_logging_dir() 0 6 2
A get_processing_worker_logging_file_path() 0 2 1
A get_cache_locked_pages_logging_file_path() 0 2 1
A get_processing_job_logging_file_path() 0 2 1
A get_processing_server_logging_file_path() 0 2 1
A get_mets_server_logging_file_path() 0 2 1
1
from pathlib import Path
2
from ocrd_utils import safe_filename, config
3
4
5
OCRD_NETWORK_MODULES = [
6
    "mets_servers",
7
    "processing_jobs",
8
    "processing_servers",
9
    "processing_workers",
10
    "processor_servers"
11
]
12
13
14
def get_root_logging_dir(module_name: str) -> Path:
15
    if module_name not in OCRD_NETWORK_MODULES:
16
        raise ValueError(f"Invalid module name: {module_name}, should be one of: {OCRD_NETWORK_MODULES}")
17
    module_log_dir = Path(config.OCRD_NETWORK_LOGS_ROOT_DIR, module_name)
18
    module_log_dir.mkdir(parents=True, exist_ok=True)
19
    return module_log_dir
20
21
22
def get_cache_locked_pages_logging_file_path() -> Path:
23
    return get_root_logging_dir("processing_servers") / "cache_locked_pages.log"
24
25
26
def get_cache_processing_requests_logging_file_path() -> Path:
27
    return get_root_logging_dir("processing_servers") / "cache_processing_requests.log"
28
29
30
def get_processing_job_logging_file_path(job_id: str) -> Path:
31
    return get_root_logging_dir("processing_jobs") / f"{job_id}.log"
32
33
34
def get_processing_server_logging_file_path(pid: int) -> Path:
35
    return get_root_logging_dir("processing_servers") / f"server.{pid}.log"
36
37
38
def get_processing_worker_logging_file_path(processor_name: str, pid: int) -> Path:
39
    return get_root_logging_dir("processing_workers") / f"worker.{pid}.{processor_name}.log"
40
41
42
def get_processor_server_logging_file_path(processor_name: str, pid: int) -> Path:
43
    return get_root_logging_dir("processor_servers") / f"server.{pid}.{processor_name}.log"
44
45
46
def get_mets_server_logging_file_path(mets_path: str) -> Path:
47
    return get_root_logging_dir("mets_servers") / f"{safe_filename(mets_path)}.log"
48