0
# Status Monitoring
1

2
Status tracking for asynchronous operations including polling, completion checking, and result analysis. Status objects track the progress and results of TAXII operations like adding objects to collections.
3

4
## Capabilities
5

6
### Status Connection
7

8
Connect to a TAXII status endpoint to monitor asynchronous operations.
9

10
```python { .api }
11
class Status:
12
    def __init__(self, url, conn=None, user=None, password=None, verify=True,
13
                 proxies=None, status_info=None, auth=None, cert=None):
14
        """
15
        Create a TAXII status endpoint connection.
16
        
17
        Parameters:
18
        - url (str): URL of TAXII status endpoint
19
        - conn (_HTTPConnection, optional): Reuse existing connection
20
        - user (str, optional): Username for HTTP basic authentication
21
        - password (str, optional): Password for HTTP basic authentication
22
        - verify (bool): Validate SSL certificates (default: True)
23
        - proxies (dict, optional): HTTP/HTTPS proxy settings
24
        - status_info (dict, optional): Pre-loaded status information
25
        - auth (requests.auth.AuthBase, optional): Custom authentication object
26
        - cert (str or tuple, optional): SSL client certificate path or (cert, key) tuple
27
        """
28
```
29

30
### Status Information
31

32
Access detailed status information about asynchronous operations.
33

34
```python { .api }
35
@property
36
def id(self) -> str:
37
    """Status operation identifier (required)."""
38

39
@property
40
def status(self) -> str:
41
    """Current status: 'pending', 'complete' (required)."""
42

43
@property
44
def request_timestamp(self) -> str:
45
    """ISO 8601 timestamp when request was initiated (optional)."""
46

47
@property
48
def total_count(self) -> int:
49
    """Total number of objects in the operation (required)."""
50

51
@property
52
def success_count(self) -> int:
53
    """Number of objects successfully processed (required)."""
54

55
@property
56
def failure_count(self) -> int:
57
    """Number of objects that failed processing (required)."""
58

59
@property
60
def pending_count(self) -> int:
61
    """Number of objects still being processed (required)."""
62

63
@property
64
def successes(self) -> list:
65
    """List of successfully processed object identifiers (optional)."""
66

67
@property
68
def failures(self) -> list:
69
    """List of failed object details with error information (optional)."""
70

71
@property
72
def pendings(self) -> list:
73
    """List of pending object identifiers (optional)."""
74

75
@property
76
def custom_properties(self) -> dict:
77
    """Custom status properties not defined in TAXII spec."""
78

79
@property
80
def _raw(self) -> dict:
81
    """Raw status response (parsed JSON)."""
82
```
83

84
### Status Operations
85

86
Monitor and update status information for asynchronous operations.
87

88
```python { .api }
89
def refresh(self, accept=None) -> None:
90
    """
91
    Update status information from the server.
92
    
93
    Parameters:
94
    - accept (str, optional): Media type for Accept header
95
    """
96

97
def wait_until_final(self, poll_interval=1, timeout=60) -> None:
98
    """
99
    Poll the status endpoint until operation completes or times out.
100
    
101
    Parameters:
102
    - poll_interval (int): Seconds between polling requests (default: 1)
103
    - timeout (int): Maximum seconds to wait (default: 60, <= 0 for no limit)
104
    """
105

106
def __bool__(self) -> bool:
107
    """
108
    Check if operation completed successfully.
109
    
110
    Returns:
111
    bool: True if status is 'complete', False otherwise
112
    """
113

114
def close(self) -> None:
115
    """Close the status connection."""
116

117
def __enter__(self):
118
    """Context manager entry."""
119

120
def __exit__(self, exc_type, exc_val, exc_tb):
121
    """Context manager exit."""
122
```
123

124
## Usage Examples
125

126
### Basic Status Monitoring
127

128
```python
129
from taxii2client import Collection
130

131
# Add objects to collection (returns Status object)
132
collection = Collection("https://taxii-server.example.com/taxii2/api1/collections/indicators/")
133
stix_envelope = {
134
    "objects": [
135
        {
136
            "type": "indicator",
137
            "id": "indicator--12345678-1234-5678-9012-123456789012",
138
            "created": "2023-01-01T00:00:00.000Z",
139
            "modified": "2023-01-01T00:00:00.000Z",
140
            "pattern": "[file:hashes.MD5 = 'd41d8cd98f00b204e9800998ecf8427e']",
141
            "labels": ["malicious-activity"]
142
        }
143
    ]
144
}
145

146
# Start async operation (don't wait for completion)
147
status = collection.add_objects(stix_envelope, wait_for_completion=False)
148

149
print(f"Operation started:")
150
print(f"  Status ID: {status.id}")
151
print(f"  Initial status: {status.status}")
152
print(f"  Total objects: {status.total_count}")
153
```
154

155
### Manual Status Polling
156

157
```python
158
import time
159

160
# Poll status manually
161
while status.status != "complete":
162
    print(f"Status: {status.status}")
163
    print(f"  Success: {status.success_count}/{status.total_count}")
164
    print(f"  Failures: {status.failure_count}")
165
    print(f"  Pending: {status.pending_count}")
166
    
167
    time.sleep(2)  # Wait 2 seconds
168
    status.refresh()  # Update status from server
169

170
print("Operation completed!")
171
```
172

173
### Automatic Status Polling
174

175
```python
176
# Use built-in polling with custom timeout
177
status = collection.add_objects(stix_envelope, wait_for_completion=False)
178

179
print(f"Starting operation: {status.id}")
180
print(f"Initial status: {status.status}")
181

182
# Wait up to 5 minutes, polling every 5 seconds
183
status.wait_until_final(poll_interval=5, timeout=300)
184

185
if status.status == "complete":
186
    print("Operation completed successfully!")
187
else:
188
    print(f"Operation timed out with status: {status.status}")
189
```
190

191
### Detailed Status Analysis
192

193
```python
194
# Analyze operation results
195
print(f"Operation Summary:")
196
print(f"  Status: {status.status}")
197
print(f"  Started: {status.request_timestamp}")
198
print(f"  Total: {status.total_count}")
199
print(f"  Successful: {status.success_count}")
200
print(f"  Failed: {status.failure_count}")
201
print(f"  Pending: {status.pending_count}")
202

203
# Check individual successes
204
if status.successes:
205
    print(f"\nSuccessful objects ({len(status.successes)}):")
206
    for success in status.successes:
207
        if isinstance(success, dict):
208
            print(f"  {success.get('id', 'Unknown ID')}")
209
        else:
210
            print(f"  {success}")
211

212
# Check individual failures
213
if status.failures:
214
    print(f"\nFailed objects ({len(status.failures)}):")
215
    for failure in status.failures:
216
        if isinstance(failure, dict):
217
            obj_id = failure.get('id', 'Unknown ID')
218
            message = failure.get('message', 'No error message')
219
            print(f"  {obj_id}: {message}")
220
        else:
221
            print(f"  {failure}")
222

223
# Check pending objects
224
if status.pendings:
225
    print(f"\nPending objects ({len(status.pendings)}):")
226
    for pending in status.pendings:
227
        if isinstance(pending, dict):
228
            print(f"  {pending.get('id', 'Unknown ID')}")
229
        else:
230
            print(f"  {pending}")
231
```
232

233
### Boolean Status Check
234

235
```python
236
# Use status as boolean (True if complete, False otherwise)
237
if status:
238
    print("Operation completed successfully")
239
else:
240
    print(f"Operation not complete: {status.status}")
241
    
242
    # Check specific completion state
243
    if status.status == "pending":
244
        print("Still processing...")
245
    elif status.failure_count > 0:
246
        print(f"Some objects failed: {status.failure_count}/{status.total_count}")
247
```
248

249
### Direct Status Retrieval
250

251
```python
252
from taxii2client import ApiRoot, Status
253

254
# Get status directly from API root
255
api_root = ApiRoot("https://taxii-server.example.com/taxii2/api1/")
256
status_id = "12345678-abcd-efgh-ijkl-123456789012"
257

258
# Retrieve existing status
259
status = api_root.get_status(status_id)
260
print(f"Retrieved status: {status.status}")
261

262
# Or create Status object directly
263
status_url = f"https://taxii-server.example.com/taxii2/api1/status/{status_id}/"
264
status = Status(status_url, user="username", password="password")
265
print(f"Direct status: {status.status}")
266
```
267

268
### Status Error Handling
269

270
```python
271
from taxii2client.exceptions import ValidationError, TAXIIServiceException
272

273
try:
274
    # Refresh status information
275
    status.refresh()
276
    
277
    # Check for validation errors in status data
278
    if status.total_count != (status.success_count + status.failure_count + status.pending_count):
279
        print("Warning: Status counts don't add up correctly")
280
        
281
except ValidationError as e:
282
    print(f"Status validation error: {e}")
283
except TAXIIServiceException as e:
284
    print(f"TAXII service error: {e}")
285
except Exception as e:
286
    print(f"Unexpected error: {e}")
287
```
288

289
### Long-Running Operations
290

291
```python
292
import time
293
from datetime import datetime, timedelta
294

295
# Start long-running operation
296
status = collection.add_objects(large_stix_envelope, wait_for_completion=False)
297
start_time = datetime.now()
298

299
print(f"Started operation at {start_time}")
300
print(f"Status ID: {status.id}")
301

302
# Monitor with progress updates
303
last_update = datetime.now()
304
while status.status != "complete":
305
    current_time = datetime.now()
306
    elapsed = current_time - start_time
307
    
308
    # Refresh status
309
    status.refresh()
310
    
311
    # Show progress every 30 seconds
312
    if current_time - last_update > timedelta(seconds=30):
313
        progress = (status.success_count + status.failure_count) / status.total_count * 100
314
        print(f"Progress: {progress:.1f}% ({elapsed.total_seconds():.0f}s elapsed)")
315
        print(f"  Processed: {status.success_count + status.failure_count}/{status.total_count}")
316
        print(f"  Success rate: {status.success_count/(status.success_count + status.failure_count)*100:.1f}%")
317
        last_update = current_time
318
    
319
    time.sleep(5)  # Check every 5 seconds
320

321
total_time = datetime.now() - start_time
322
print(f"Operation completed in {total_time.total_seconds():.0f} seconds")
323
```
324

325
### Context Manager Usage
326

327
```python
328
# Direct status connection with automatic cleanup
329
status_url = "https://taxii-server.example.com/taxii2/api1/status/12345/"
330
with Status(status_url, user="user", password="pass") as status:
331
    print(f"Status: {status.status}")
332
    print(f"Success: {status.success_count}/{status.total_count}")
333
    # Connection automatically closed when exiting context
334
```