pycti.api.opencti_api_work ========================== .. py:module:: pycti.api.opencti_api_work Classes ------- .. autoapisummary:: pycti.api.opencti_api_work.OpenCTIApiWork Module Contents --------------- .. py:class:: OpenCTIApiWork(api) OpenCTI Work API class. Manages work/job operations for connectors. :param api: instance of :py:class:`~pycti.api.opencti_api_client.OpenCTIApiClient` :type api: OpenCTIApiClient Initialize the OpenCTIApiWork instance. :param api: OpenCTI API client instance :type api: OpenCTIApiClient .. py:attribute:: api .. py:method:: to_received(work_id: str, message: str) Mark work as received. :param work_id: the work id :type work_id: str :param message: the message to report :type message: str :return: None :rtype: None .. py:method:: to_processed(work_id: str, message: str, in_error: bool = False) Mark work as processed. :param work_id: the work id :type work_id: str :param message: the message to report :type message: str :param in_error: whether the work completed with error, defaults to False :type in_error: bool, optional :return: None :rtype: None .. py:method:: ping(work_id: str) Ping a work to keep it alive. :param work_id: the work id :type work_id: str :return: None :rtype: None .. py:method:: report_expectation(work_id: str, error) Report a work expectation. :param work_id: the work id :type work_id: str :param error: the error to report (WorkErrorInput format) :type error: dict :return: None :rtype: None .. py:method:: add_expectations(work_id: str, expectations: int) Add expectations to a work. :param work_id: the work id :type work_id: str :param expectations: the number of expectations to add :type expectations: int :return: whether the work is still alive :rtype: bool .. py:method:: add_draft_context(work_id: str, draft_context: str) Add draft context to a work. :param work_id: the work id :type work_id: str :param draft_context: the draft context to add :type draft_context: str :return: None :rtype: None .. py:method:: initiate_work(connector_id: str, friendly_name: str, is_multipart: bool = False) -> Optional[str] Initiate a new work for a connector. :param connector_id: the connector id :type connector_id: str :param friendly_name: the friendly name for the work :type friendly_name: str :param is_multipart: indicates whether multiple calls to `add_expectations` are to be expected during the lifetime of the work. In consequence the work won't automatically transition to `complete` when the number of calls to `report_expectation` matches the expectations but only when an explicit call to `to_processed` is made. Should be set to `True` when sending multiple STIX bundles consecutively via `send_stix2_bundle` during the work's lifetime. Defaults to `False`. :type is_multipart: bool :return: the work id or None if bundle_send_to_queue is False :rtype: str or None .. py:method:: delete_work(work_id: str) Delete a work. .. deprecated:: Use :meth:`delete` instead. :param work_id: the work id :type work_id: str :return: the response data :rtype: dict .. py:method:: delete(**kwargs) Delete a work by id. :param id: the work id :type id: str :return: the response data :rtype: dict or None .. py:method:: wait_for_work_to_finish(work_id: str) Wait for a work to finish. :param work_id: the work id :type work_id: str :return: empty string if error, None otherwise :rtype: str or None .. py:method:: get_work(work_id: str) -> Dict Get a work by id. :param work_id: the work id :type work_id: str :return: the work data :rtype: dict .. py:method:: get_is_work_alive(work_id: str) -> bool Check if a work is alive. :param work_id: the work id :type work_id: str :return: whether the work is alive :rtype: bool .. py:method:: get_connector_works(connector_id: str) -> List[Dict] Get all works for a connector. :param connector_id: the connector id :type connector_id: str :return: list of work dictionaries sorted by timestamp :rtype: list[dict]