Function get_timeseries

sdk/python/feast/api/registry/rest/monitoring.py:370–397

(
        project: str = Query(...),
        feature_view_name: Optional[str] = Query(None),
        feature_name: Optional[str] = Query(None),
        feature_service_name: Optional[str] = Query(None),
        granularity: Optional[str] = Query(None),
        data_source_type: Optional[str] = Query(None),
        start_date: Optional[str] = Query(None),
        end_date: Optional[str] = Query(None),
    )

    @router.get("/monitoring/metrics/timeseries", tags=["Monitoring"])
    async def get_timeseries(
        project: str = Query(...),
        feature_view_name: Optional[str] = Query(None),
        feature_name: Optional[str] = Query(None),
        feature_service_name: Optional[str] = Query(None),
        granularity: Optional[str] = Query(None),
        data_source_type: Optional[str] = Query(None),
        start_date: Optional[str] = Query(None),
        end_date: Optional[str] = Query(None),
    ):
        store = _get_store()
        if feature_view_name:
            fv = store.registry.get_feature_view(
                name=feature_view_name, project=project
            )
            assert_permissions(fv, actions=[AuthzedAction.DESCRIBE])

        svc = _get_monitoring_service()
        return svc.get_timeseries(
            project=project,
            feature_view_name=feature_view_name,
            feature_name=feature_name,
            feature_service_name=feature_service_name,
            granularity=granularity,
            data_source_type=data_source_type,
            start_date=date.fromisoformat(start_date) if start_date else None,
            end_date=date.fromisoformat(end_date) if end_date else None,
        )

    return router