get_context_graph ¶

def get_context_graph(
    self,
    start_time: datetime.datetime,
    end_time: datetime.datetime,
    *,
    config_overrides: Optional[CloudWatchOmniClientConfig] = None,
    node_filters: Optional[
        "capo_cloudwatchomni.types.node_filters.NodeFilters"
    ] = None,
    edge_filters: Optional[
        "capo_cloudwatchomni.types.edge_filters.EdgeFilters"
    ] = None,
    depth: Optional[int] = None,
    max_results: Optional[int] = None,
    max_edges_per_node: Optional[int] = None,
    include_metadata: Optional[bool] = None,
    next_token: Optional[
        "capo_cloudwatchomni.types.pagination_token.PaginationToken"
    ] = None,
) -> "capo_cloudwatchomni.types.get_context_graph_output.GetContextGraphOutput":
    """Queries the context graph with filtering, traversal, and pagination support. Pagination note: nodes and edges are returned together as a coherent subgraph. Pagination cursors advance over nodes (the primary collection); each page includes all edges connecting nodes within that page. Callers should treat nodes as the paginated collection and edges as supplementary relationship data attached to those nodes.

    Args:
        node_filters: Criteria restricting which nodes are returned.
        edge_filters: Criteria restricting which edges are returned.
        start_time: Start of the time range (UTC), inclusive.
        end_time: End of the time range (UTC), inclusive.
        depth: How many hops to traverse out from the nodes matched by nodeFilters. 0 returns only the matched nodes themselves.
        max_results: The maximum number of nodes to return in a single page.
        max_edges_per_node: The maximum number of edges to return per node, bounding the fan-out of a densely connected node.
        include_metadata: Whether to return the metadata block, semantics included, on each node and edge. Off by default because it costs an extra lookup per returned node.
        next_token: Pagination token from a previous response, to retrieve the next page.

    Raises:
        capo_cloudwatchomni.errors.access_denied_exception.AccessDeniedException: The caller is not authorized to perform this action.
        capo_cloudwatchomni.errors.internal_server_exception.InternalServerException: An unexpected error occurred while processing the request.
        capo_cloudwatchomni.errors.throttling_exception.ThrottlingException: The request was throttled due to exceeding the allowed request rate.
        capo_cloudwatchomni.errors.validation_exception.ValidationException: A parameter is specified incorrectly.
        capo_cloudwatchomni.errors.UnknownServiceError: The service returned an error code this client does not model.

    Examples:
        Query a service and its immediate dependencies
        The following example returns context graph nodes matching the filter and traverses one hop out to their direct dependencies, over a one-hour window. Payloads are shown as JSON; on the wire they are CBOR-encoded.

        >>> client.get_context_graph(node_filters={'nodeType': 'SERVICE', 'namespace': ['ecommerce']}, start_time='2026-09-16T00:00:00Z', end_time='2026-09-16T01:00:00Z', depth=1, max_results=100, include_metadata=False)
    """

    def _handler(
        req: "OperationRequest[capo_cloudwatchomni.types.get_context_graph_input.GetContextGraphInput]",
    ) -> OperationResponse[
        "capo_cloudwatchomni.types.get_context_graph_output.GetContextGraphOutput"
    ]:
        import capo_cloudwatchomni._operations.cloud_watch_omni_frontend.get_context_graph

        output, http_response = (
            capo_cloudwatchomni._operations.cloud_watch_omni_frontend.get_context_graph.get_context_graph(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("GetContextGraph")
    )
    input_: capo_cloudwatchomni.types.get_context_graph_input.GetContextGraphInput = {
        "start_time": start_time,
        "end_time": end_time,
    }
    if node_filters is not None:
        input_["node_filters"] = node_filters
    if edge_filters is not None:
        input_["edge_filters"] = edge_filters
    if depth is not None:
        input_["depth"] = depth
    if max_results is not None:
        input_["max_results"] = max_results
    if max_edges_per_node is not None:
        input_["max_edges_per_node"] = max_edges_per_node
    if include_metadata is not None:
        input_["include_metadata"] = include_metadata
    if next_token is not None:
        input_["next_token"] = next_token

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    response.response.close()
    return response.output