pycti.api.opencti_api_work

Classes

OpenCTIApiWork

OpenCTI Work API class.

Module Contents

class pycti.api.opencti_api_work.OpenCTIApiWork(api)[source]

OpenCTI Work API class.

Manages work/job operations for connectors.

Parameters:

api (OpenCTIApiClient) – instance of OpenCTIApiClient

Initialize the OpenCTIApiWork instance.

Parameters:

api (OpenCTIApiClient) – OpenCTI API client instance

api[source]
to_received(work_id: str, message: str)[source]

Mark work as received.

Parameters:

  • work_id (str) – the work id

  • message (str) – the message to report

Returns:

None

Return type:

None

to_processed(work_id: str, message: str, in_error: bool = False)[source]

Mark work as processed.

Parameters:

  • work_id (str) – the work id

  • message (str) – the message to report

  • in_error (bool, optional) – whether the work completed with error, defaults to False

Returns:

None

Return type:

None

ping(work_id: str)[source]

Ping a work to keep it alive.

Parameters:

work_id (str) – the work id

Returns:

None

Return type:

None

report_expectation(work_id: str, error)[source]

Report a work expectation.

Parameters:

  • work_id (str) – the work id

  • error (dict) – the error to report (WorkErrorInput format)

Returns:

None

Return type:

None

add_expectations(work_id: str, expectations: int)[source]

Add expectations to a work.

Parameters:

  • work_id (str) – the work id

  • expectations (int) – the number of expectations to add

Returns:

whether the work is still alive

Return type:

bool

add_draft_context(work_id: str, draft_context: str)[source]

Add draft context to a work.

Parameters:

  • work_id (str) – the work id

  • draft_context (str) – the draft context to add

Returns:

None

Return type:

None

initiate_work(connector_id: str, friendly_name: str, is_multipart: bool = False) → str | None[source]

Initiate a new work for a connector.

Parameters:

  • connector_id (str) – the connector id

  • friendly_name (str) – the friendly name for the work

  • is_multipart (bool) – 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.

Returns:

the work id or None if bundle_send_to_queue is False

Return type:

str or None

delete_work(work_id: str)[source]

Delete a work.

Deprecated since version Use: delete() instead.

Parameters:

work_id (str) – the work id

Returns:

the response data

Return type:

dict

delete(**kwargs)[source]

Delete a work by id.

Parameters:

id (str) – the work id

Returns:

the response data

Return type:

dict or None

wait_for_work_to_finish(work_id: str)[source]

Wait for a work to finish.

Parameters:

work_id (str) – the work id

Returns:

empty string if error, None otherwise

Return type:

str or None

get_work(work_id: str) → Dict[source]

Get a work by id.

Parameters:

work_id (str) – the work id

Returns:

the work data

Return type:

dict

get_is_work_alive(work_id: str) → bool[source]

Check if a work is alive.

Parameters:

work_id (str) – the work id

Returns:

whether the work is alive

Return type:

bool

get_connector_works(connector_id: str) → List[Dict][source]

Get all works for a connector.

Parameters:

connector_id (str) – the connector id

Returns:

list of work dictionaries sorted by timestamp

Return type:

list[dict]