Service bookkeeper

bookkeeper.py

The bookkeeper service of mercure, which receives notifications from all mercure services and stores the information in a database.

class bookkeeping.bookkeeper.TokenAuth[source]

Bases: BaseTokenAuth

async verify(token: str)[source]

bookkeeping.bookkeeper.app: Starlette

bookkeeping.bookkeeper.create_app() → Starlette[source]

bookkeeping.bookkeeper.lifespan(app)[source]

bookkeeping.bookkeeper.main(args=['-b', 'html', '-d', '_build/doctrees', '.', '_build/html']) → None[source]

bookkeeping.bookkeeper.migrate_database() → None[source]

async bookkeeping.bookkeeper.parse_and_submit_tags(payload) → None[source]: Helper function that reads series information from the request body.

async bookkeeping.bookkeeper.post_mercure_event(request) → JSONResponse[source]: Endpoint for receiving mercure system events.

async bookkeeping.bookkeeper.post_task_event(request) → JSONResponse[source]: Endpoint for logging all events related to one series.

async bookkeeping.bookkeeper.post_webgui_event(request) → JSONResponse[source]: Endpoint for logging relevant events of the webgui.

async bookkeeping.bookkeeper.processor_logs(request) → JSONResponse[source]: Endpoint for receiving mercure system events.

async bookkeeping.bookkeeper.register_dicom(request) → JSONResponse[source]: Endpoint for registering newly received DICOM files. Called by the getdcmtags module.

async bookkeeping.bookkeeper.register_series(request) → JSONResponse[source]: Endpoint that is called by the router whenever a new series arrives.

async bookkeeping.bookkeeper.register_task(request) → JSONResponse[source]

async bookkeeping.bookkeeper.server_error(request, exc) → Response[source]: Return an HTTP 500 page.

async bookkeeping.bookkeeper.store_processor_output(request) → JSONResponse[source]

async bookkeeping.bookkeeper.test_begin(request) → JSONResponse[source]

async bookkeeping.bookkeeper.test_end(request) → JSONResponse[source]

async bookkeeping.bookkeeper.test_endpoint(request) → JSONResponse[source]: Endpoint for testing that the bookkeeper is active.

async bookkeeping.bookkeeper.update_task(request) → JSONResponse[source]: Endpoint that is called by the router whenever a new series arrives.

config.py

Handling of configuration setting for the bookkeeper service

bookkeeping.config.API_KEY: Optional[str]

bookkeeping.config.BOOKKEEPER_HOST: str

bookkeeping.config.BOOKKEEPER_PORT: int

bookkeeping.config.DATABASE_SCHEMA: Optional[str]

bookkeeping.config.DATABASE_URL: str

bookkeeping.config.DEBUG_MODE: bool

bookkeeping.config.bookkeeper_config: Config

bookkeeping.config.read_bookkeeper_config() → Config[source]

bookkeeping.config.set_api_key() → None[source]

database.py

Database functions needed for the bookkeeper service.

bookkeeping.database.database: Database

bookkeeping.database.dicom_files: Table

bookkeeping.database.dicom_series: Table

bookkeeping.database.dicom_series_map: Table

bookkeeping.database.file_events: Table

bookkeeping.database.init_database(url=None, schema=None) → Database[source]

bookkeeping.database.mercure_events: Table

bookkeeping.database.metadata: MetaData

bookkeeping.database.processor_logs_table: Table

bookkeeping.database.processor_outputs_table: Table

bookkeeping.database.series_sequence_data: Table

bookkeeping.database.task_events: Table

bookkeeping.database.tasks_table: Table

bookkeeping.database.tests_table: Table

bookkeeping.database.webgui_events: Table

helper.py

Helper functions for the bookkeeper service.

class bookkeeping.helper.CustomJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

default(obj)[source]

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class bookkeeping.helper.CustomJSONResponse(content: Any, status_code: int = 200, headers: Optional[Mapping[str, str]] = None, media_type: Optional[str] = None, background: Optional[BackgroundTask] = None)[source]

Bases: JSONResponse

render(content: Any) → bytes[source]

query.py

Entry functions of the bookkeeper for querying processing information.

bookkeeping.query.convert_key(tag_key)[source]

bookkeeping.query.convert_to_serializable(obj)[source]: Converts non-serializable objects to serializable types.

bookkeeping.query.dicom_to_readable_json(ds: Dataset)[source]

Converts a DICOM file to a human-readable JSON format.

Args:: file_path (str): Path to the DICOM file. output_file_path (str): Path to save the JSON output.

async bookkeeping.query.find_task(request) → JSONResponse[source]

async bookkeeping.query.get_dicom_files(request) → JSONResponse[source]: Endpoint for getting all events related to one series.

async bookkeeping.query.get_series(request) → JSONResponse[source]: Endpoint for retrieving series in the database.

async bookkeeping.query.get_task_events(request) → JSONResponse[source]: Endpoint for getting all events related to one task.

async bookkeeping.query.get_task_info(request) → JSONResponse[source]

async bookkeeping.query.get_task_process_logs(request) → JSONResponse[source]: Endpoint for getting all processing logs related to one series.

async bookkeeping.query.get_task_process_results(request) → JSONResponse[source]: Endpoint for getting all processing results from a task.

async bookkeeping.query.get_tasks(request) → JSONResponse[source]: Endpoint for retrieving tasks in the database.

async bookkeeping.query.get_test_task(request) → JSONResponse[source]

bookkeeping.query.set_timezone_conversion() → None[source]