Skip to content

onconova.core.healthcheck

This module provides an API endpoint for health checking the server and its database.

HealthCheck

Bases: Schema

Schema representing the health status of the server and its components.

Attributes:

Name Type Description
server Literal['ok']

Indicates whether the server is online.
database Union[Literal['ok'], Literal['error']]

Indicates the status of the database connection.
database_connection_time_ms Optional[float]

Time taken to connect to the database in milliseconds.
migrations Union[Literal['ok'], Literal['pending'], Literal['error']]

Status of database migrations.

database class-attribute instance-attribute

database_connection_time_ms class-attribute instance-attribute

migrations class-attribute instance-attribute

server class-attribute instance-attribute

HealthCheckController

Bases: ControllerBase

API controller for performing health checks on the server and its database.

health_check()

Performs a health check of the server, database connection, and database migrations.

Source code in onconova/core/healthcheck.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@route.get(
    path="",
    response={
        200: HealthCheck,
        401: None,
        400: None,
        403: None,
        500: None,
    },
    operation_id="healthcheck",
    openapi_extra=dict(security=[]),
)
def health_check(self):
    """
    Performs a health check of the server, database connection, and database migrations.
    """
    # Check server status (if this endpoint is hit, server is up)
    server_status = "ok"
    # Check database connection and measure connection speed
    db_connection_time_ms = None
    try:
        start = time.time()
        connections["default"].cursor()
        end = time.time()
        database_status = "ok"
        db_connection_time_ms = (end - start) * 1000  # milliseconds
    except OperationalError:
        database_status = "error"

    # Check for unapplied migrations
    try:
        connection = connections[DEFAULT_DB_ALIAS]
        executor = MigrationExecutor(connection)
        plan = executor.migration_plan(executor.loader.graph.leaf_nodes())
        if plan:
            migrations_status = "pending"
        else:
            migrations_status = "ok"
    except (OperationalError, ProgrammingError):
        migrations_status = "error"

    return HealthCheck(
        server=server_status,
        database=database_status,
        database_connection_time_ms=(
            round(db_connection_time_ms, 2)
            if db_connection_time_ms is not None
            else None
        ),
        migrations=migrations_status,
    )
runner