Source code for britecore_sdk.api.api_calls.v2.notes

"""BriteCore v2 Notes API endpoint wrappers.

Provides:
    retrieve_notes -- Retrieve paginated, filterable notes for any entity ID.
"""

from json import loads
from logging import Logger
from typing import Any, Unpack

from urllib3 import BaseHTTPResponse, HTTPResponse, Timeout

from britecore_sdk import logger
from britecore_sdk.api.api_calls import (
    BritecoreAPIClient,
    RequestParameters,
    api_client,
    web_timeout_long,
)
from britecore_sdk.api.api_calls.v2._common import post as common_post

LOGGER: Logger = logger

API_CLIENT: BritecoreAPIClient = api_client


[docs] def new_note( payload: dict[str, Any] | None = None, **kwargs: Unpack[RequestParameters], ) -> Any: """Create a new note record for an entity.""" return common_post("/api/v2/notes/newNote", payload=payload, **kwargs)
[docs] def store_note( payload: dict[str, Any] | None = None, **kwargs: Unpack[RequestParameters], ) -> Any: """Persist updates for an existing note record.""" return common_post("/api/v2/notes/storeNote", payload=payload, **kwargs)
[docs] def retrieve_notes( entity_id: str, pageSize: int | None = 1000, searchString: str | None = None, aggregateAll: bool | None = False, advSearch: bool | None = False, filterAlerts: bool | None = False, filterUserGen: bool | None = False, orderBy: str | None = "", page: int | None = 0, ascending: bool | None = False, note_type: str | None = "", filterExcludeAlerts: bool | None = False, filterSystemNotesOnly: bool | None = False, **kwargs: Unpack[RequestParameters], ) -> Any: """Retrieve notes for an entity with filtering and pagination controls. POST /api/v2/notes/retrieveNotes """ LOGGER.debug("Getting notes") notes_json: dict[str, Any] = {} local_env: dict[str, str | None] = {**locals()} for _, (k, v) in enumerate( local_env.items() ): # Add any non-default parameters to the request if v: if k == "note_type": k = "type" if k == "entity_id": k = "id" notes_json.update({k: v}) provided_timeout: Timeout | None = kwargs.get("request_timeout") if not provided_timeout: kwargs.update({"request_timeout": Timeout(web_timeout_long)}) request_result: BaseHTTPResponse | HTTPResponse | None = API_CLIENT.do_request( path="/api/v2/notes/retrieveNotes", json=notes_json, **kwargs ) if not request_result: return [] try: return loads(request_result.data.decode("utf-8"))["records"] except KeyError: return []
__all__ = ["new_note", "retrieve_notes", "store_note"]