Skip to content

dp3.api.routers.root

health_check async 

health_check() -> HealthCheckResponse

Health check

Returns simple 'It works!' response.

Source code in dp3/api/routers/root.py 
21
22
23
24
25
26
27
@router.get("/", tags=["Health"])
async def health_check() -> HealthCheckResponse:
    """Health check

    Returns simple 'It works!' response.
    """
    return HealthCheckResponse()

insert_datapoints async 

insert_datapoints(dps: list[DataPoint], request: Request) -> SuccessResponse

Insert datapoints

Validates and pushes datapoints into task queue, so they are processed by one of DP3 workers.

Source code in dp3/api/routers/root.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@router.post(DATAPOINTS_INGESTION_URL_PATH, tags=["Data ingestion"])
async def insert_datapoints(dps: list[DataPoint], request: Request) -> SuccessResponse:
    """Insert datapoints

    Validates and pushes datapoints into task queue, so they are processed by one of DP3 workers.
    """
    # Convert to DP3 datapoints
    # This should not fail as all datapoints are already validated
    dp3_dps = [api_to_dp3_datapoint(dp.model_dump()) for dp in dps]

    # Group datapoints by etype-eid
    tasks_dps = defaultdict(list)
    for dp in dp3_dps:
        key = (dp.etype, dp.eid)
        tasks_dps[key].append(dp)

    # Create tasks
    tasks = []
    with task_context(MODEL_SPEC):
        for k in tasks_dps:
            etype, eid = k

            # This shouldn't fail either
            tasks.append(DataPointTask(etype=etype, eid=eid, data_points=tasks_dps[k]))

    # Push tasks to task queue
    for task in tasks:
        TASK_WRITER.put_task(task, False)

    # Log datapoints
    DP_LOGGER.log_good(dp3_dps, src=request.client.host)

    return SuccessResponse()

list_entity_types async 

list_entity_types() -> dict[str, EntityState]

List entity types

Returns dictionary containing all entity types configured -- their simplified configuration and current state information.

Source code in dp3/api/routers/root.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@router.get("/entities", tags=["Entity"])
async def list_entity_types() -> dict[str, EntityState]:
    """List entity types

    Returns dictionary containing all entity types configured -- their simplified configuration
    and current state information.
    """
    entities = {}

    for etype in MODEL_SPEC.entities:
        entity_spec = MODEL_SPEC.entity(etype)
        entities[etype] = {
            "id": etype,
            "id_data_type": entity_spec.id_data_type.root,
            "name": entity_spec.name,
            "attribs": MODEL_SPEC.attribs(etype),
            "eid_estimate_count": DB.estimate_count_eids(etype),
        }

    return entities