API Root Management

class ApiRoot:
    def __init__(self, url, conn=None, user=None, password=None, verify=True,
                 proxies=None, auth=None, cert=None):
        """
        Create a TAXII API root endpoint connection.
        
        Parameters:
        - url (str): URL of TAXII API root endpoint
        - conn (_HTTPConnection, optional): Reuse existing connection
        - user (str, optional): Username for HTTP basic authentication
        - password (str, optional): Password for HTTP basic authentication
        - verify (bool): Validate SSL certificates (default: True)
        - proxies (dict, optional): HTTP/HTTPS proxy settings
        - auth (requests.auth.AuthBase, optional): Custom authentication object
        - cert (str or tuple, optional): SSL client certificate path or (cert, key) tuple
        """

@property
def title(self) -> str:
    """API root title (required)."""

@property
def description(self) -> str:
    """API root description (optional)."""

@property
def versions(self) -> list[str]:
    """List of supported TAXII versions (required)."""

@property
def max_content_length(self) -> int:
    """Maximum content length in bytes for requests (required)."""

@property
def custom_properties(self) -> dict:
    """Custom API root properties not defined in TAXII spec."""

@property
def _raw(self) -> dict:
    """Raw API root information response (parsed JSON)."""

@property
def collections(self) -> list[Collection]:
    """List of Collection instances available in this API root."""

def refresh(self, accept=None) -> None:
    """
    Update API root information and list of collections.
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    """

def refresh_information(self, accept=None) -> None:
    """
    Update only the API root properties (not collections).
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    """

def refresh_collections(self, accept=None) -> None:
    """
    Update only the list of collections (not API root info).
    
    Parameters:
    - accept (str, optional): Media type for Accept header
    """

def get_status(self, status_id, accept=None) -> Status:
    """
    Retrieve status information for an asynchronous operation.
    
    Parameters:
    - status_id (str): Status identifier from previous operation
    - accept (str, optional): Media type for Accept header
    
    Returns:
    Status: Status object with operation details
    """

def close(self) -> None:
    """Close the API root connection."""

def __enter__(self):
    """Context manager entry."""

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""

from taxii2client import Server

# Get API root from server
server = Server("https://taxii-server.example.com/taxii2/")
api_root = server.default or server.api_roots[0]

print(f"API Root: {api_root.title}")
print(f"Description: {api_root.description}")
print(f"Supported versions: {', '.join(api_root.versions)}")
print(f"Max content length: {api_root.max_content_length} bytes")

from taxii2client import ApiRoot

# Connect directly to specific API root
api_root = ApiRoot(
    url="https://taxii-server.example.com/taxii2/api1/",
    user="username",
    password="password"
)

# Access API root information
print(f"Title: {api_root.title}")
print(f"Versions: {api_root.versions}")

# List all collections in API root
print(f"Collections ({len(api_root.collections)}):")
for collection in api_root.collections:
    print(f"  - {collection.title} (ID: {collection.id})")
    print(f"    Can Read: {collection.can_read}")
    print(f"    Can Write: {collection.can_write}")
    print(f"    Media Types: {', '.join(collection.media_types)}")

# Find collection by ID
target_id = "malware-indicators"
collection = next(
    (c for c in api_root.collections if c.id == target_id), 
    None
)
if collection:
    print(f"Found collection: {collection.title}")
else:
    print(f"Collection '{target_id}' not found")

# Find collections by capability
readable_collections = [c for c in api_root.collections if c.can_read]
writable_collections = [c for c in api_root.collections if c.can_write]

print(f"Readable collections: {len(readable_collections)}")
print(f"Writable collections: {len(writable_collections)}")

# Monitor status of previous operation
status_id = "status-12345-abcde"
status = api_root.get_status(status_id)

print(f"Status ID: {status.id}")
print(f"Status: {status.status}")
print(f"Total: {status.total_count}")
print(f"Success: {status.success_count}")
print(f"Failures: {status.failure_count}")
print(f"Pending: {status.pending_count}")

# Check if operation completed successfully
if status:  # Uses __bool__ method
    print("Operation completed successfully")
else:
    print(f"Operation not complete: {status.status}")

# Refresh all information
api_root.refresh()

# Refresh only API root info (faster)
api_root.refresh_information()

# Refresh only collections (useful if collections change frequently)
api_root.refresh_collections()

# Check for new collections after refresh
print(f"Collections after refresh: {len(api_root.collections)}")

# Automatically handle connection cleanup
api_root_url = "https://taxii-server.example.com/taxii2/api1/"
with ApiRoot(api_root_url, user="user", password="pass") as api_root:
    print(f"Connected to: {api_root.title}")
    for collection in api_root.collections:
        print(f"Collection: {collection.title}")
    # Connection automatically closed when exiting context

# Access custom properties not in TAXII spec
if api_root.custom_properties:
    print("Custom properties:")
    for key, value in api_root.custom_properties.items():
        print(f"  {key}: {value}")

# Access raw JSON response
raw_data = api_root._raw
print(f"Raw response keys: {list(raw_data.keys())}")