Log Extraction API

Retrieve and export Silo audit logs. Supports 26 log types, pagination, date filtering, and CSV/JSON export. Requires LOG_TOKEN.

Bases: BaseAPIClient

API client for Silo log extraction operations.

This class provides methods for extracting audit logs and activity data from the Authentic8 Silo platform for compliance, monitoring, and analysis purposes.

The log extraction API allows you to: - Extract logs by type and sequence range - Retrieve comprehensive audit trails - Handle large log datasets with pagination - Filter logs by organization and time periods - Export logs for external analysis and compliance

Supported log types include: - ADMIN_AUDIT: Administrative actions and changes - AUTH: Authentication and authorization events - COOKIES: Cookie handling and management - DOWNLOAD: File download activities - UPLOAD: File upload activities - POST_DATA: Form submissions and POST requests - SESSION: Session lifecycle events - ENC: Encryption and security events - URL: URL access and navigation - BLOCKED_URL: Blocked URL attempts - LOCATION_CHANGE: Geographic location changes - TRANSLATION: Content translation events - A8SS: Silo storage system events - EXPLOIT: Exploit detection events - PRINT: Print operations - SMS: SMS-related events - ISOLATE_BYPASS: Isolation bypass events - TRAFFICMAN: Traffic management events - HARVEST: Harvesting operation logs - CASE_MANAGER: Case management events - CLIPBOARD: Clipboard operations - LAUNCHER: Launcher events - EXTENSION: Browser extension events - APP_LAUNCH: Application launch events - EVENT: Generic platform events - NEXUS: Nexus AI conversation events

Attributes:

Name	Type	Description
VALID_LOG_TYPES	set[str]	Set of 26 valid log type strings. All types have canonical schemas defined in silo_sdk.logging.log_schemas. ENC logs have wrapper fields but the enc field contains encrypted content requiring decryption.
MAX_ENC_BATCH_SIZE	int	Maximum batch size (30) for ENC log type extraction to prevent payload size issues with encrypted log entries.
EXTRACT_LOCK_TTL	int	Default request timeout (600s) for this client, matching the backend's per-extract lock TTL. Overridable with the LOG_EXTRACT_TIMEOUT config key; REQUEST_TIMEOUT does not apply to extraction.

Example

from silo_sdk import LogExtractionAPI, load_config config = load_config() logs = LogExtractionAPI(config)

Extract recent authentication logs

auth_logs = logs.extract_logs( ... org="my_organization", ... start_seq=1000, ... log_types=["AUTH", "SESSION"], ... limit=500 ... )

Extract all logs with pagination

all_logs = logs.extract_all_logs( ... org="my_organization", ... start_seq=1, ... log_types=["ADMIN_AUDIT", "AUTH"] ... )

Source code in silo_sdk/logging/extraction_api.py

class LogExtractionAPI(BaseAPIClient):
    """API client for Silo log extraction operations.

    This class provides methods for extracting audit logs and activity data from
    the Authentic8 Silo platform for compliance, monitoring, and analysis purposes.

    The log extraction API allows you to:
    - Extract logs by type and sequence range
    - Retrieve comprehensive audit trails
    - Handle large log datasets with pagination
    - Filter logs by organization and time periods
    - Export logs for external analysis and compliance

    Supported log types include:
    - ADMIN_AUDIT: Administrative actions and changes
    - AUTH: Authentication and authorization events
    - COOKIES: Cookie handling and management
    - DOWNLOAD: File download activities
    - UPLOAD: File upload activities
    - POST_DATA: Form submissions and POST requests
    - SESSION: Session lifecycle events
    - ENC: Encryption and security events
    - URL: URL access and navigation
    - BLOCKED_URL: Blocked URL attempts
    - LOCATION_CHANGE: Geographic location changes
    - TRANSLATION: Content translation events
    - A8SS: Silo storage system events
    - EXPLOIT: Exploit detection events
    - PRINT: Print operations
    - SMS: SMS-related events
    - ISOLATE_BYPASS: Isolation bypass events
    - TRAFFICMAN: Traffic management events
    - HARVEST: Harvesting operation logs
    - CASE_MANAGER: Case management events
    - CLIPBOARD: Clipboard operations
    - LAUNCHER: Launcher events
    - EXTENSION: Browser extension events
    - APP_LAUNCH: Application launch events
    - EVENT: Generic platform events
    - NEXUS: Nexus AI conversation events

    Attributes:
        VALID_LOG_TYPES: Set of 26 valid log type strings.
            All types have canonical schemas defined in
            ``silo_sdk.logging.log_schemas``. ENC logs have wrapper fields
            but the ``enc`` field contains encrypted content requiring
            decryption.
        MAX_ENC_BATCH_SIZE: Maximum batch size (30) for ENC log type extraction
            to prevent payload size issues with encrypted log entries.
        EXTRACT_LOCK_TTL: Default request timeout (600s) for this client,
            matching the backend's per-extract lock TTL. Overridable with the
            ``LOG_EXTRACT_TIMEOUT`` config key; ``REQUEST_TIMEOUT`` does not
            apply to extraction.

    Example:
        >>> from silo_sdk import LogExtractionAPI, load_config
        >>> config = load_config()
        >>> logs = LogExtractionAPI(config)
        >>>
        >>> # Extract recent authentication logs
        >>> auth_logs = logs.extract_logs(
        ...     org="my_organization",
        ...     start_seq=1000,
        ...     log_types=["AUTH", "SESSION"],
        ...     limit=500
        ... )
        >>>
        >>> # Extract all logs with pagination
        >>> all_logs = logs.extract_all_logs(
        ...     org="my_organization",
        ...     start_seq=1,
        ...     log_types=["ADMIN_AUDIT", "AUTH"]
        ... )
    """

    # Maximum batch size for ENC log type extraction
    MAX_ENC_BATCH_SIZE: int = 30

    # The backend serializes extracts behind a lock with a 600-second TTL.
    # A client timeout shorter than that abandons a request that is still
    # running server-side and leaves the lock held, so the retry fails with
    # "log.extract already in progress" until the TTL expires -- a slow
    # success becomes a ten-minute outage. Waiting the full TTL means the
    # client never gives up on work that can still land, and anything that
    # does time out has also released its lock, so the retry is clean.
    EXTRACT_LOCK_TTL: int = 600

    # Valid log types supported by the Silo platform
    VALID_LOG_TYPES: set[str] = {
        "ADMIN_AUDIT",  # Administrative actions and changes
        "AUTH",  # Authentication and authorization events
        "COOKIES",  # Cookie handling and management
        "DOWNLOAD",  # File download activities
        "UPLOAD",  # File upload activities
        "POST_DATA",  # Form submissions and POST requests
        "SESSION",  # Session lifecycle events
        "ENC",  # Encryption and security events
        "URL",  # URL access and navigation
        "BLOCKED_URL",  # Blocked URL attempts
        "LOCATION_CHANGE",  # Geographic location changes
        "TRANSLATION",  # Content translation events
        "A8SS",  # Silo storage system events
        "EXPLOIT",  # Exploit detection events
        "PRINT",  # Print operations
        "SMS",  # SMS-related events
        "ISOLATE_BYPASS",  # Isolation bypass events
        "TRAFFICMAN",  # Traffic management events
        "HARVEST",  # Harvesting operation logs
        "CASE_MANAGER",  # Case management events
        "CLIPBOARD",  # Clipboard operations
        "LAUNCHER",  # Launcher events
        "EXTENSION",  # Browser extension events
        "APP_LAUNCH",  # Application launch events
        "EVENT",  # Generic platform events
        "NEXUS",  # Nexus AI conversation events
    }

    def __init__(self, config: dict[str, Any]) -> None:
        """Initialize the log extraction API client.

        This client takes its request timeout from ``LOG_EXTRACT_TIMEOUT``
        (``A8_LOG_EXTRACT_TIMEOUT``), which defaults to 600 seconds -- the
        server's per-extract lock TTL -- rather than from the SDK-wide
        ``REQUEST_TIMEOUT`` default of 30 seconds. Extracts routinely run for
        minutes, and a client that gives up while the lock is still held
        blocks its own retries until that TTL expires.

        The 600-second default only ever raises the extraction timeout; it
        never lowers one already set higher. A ``REQUEST_TIMEOUT`` above 600
        is kept as-is, so anyone who raised it to work around the previous
        30-second limit is not cut back on upgrade. An explicit
        ``LOG_EXTRACT_TIMEOUT`` wins outright, including a value below
        ``REQUEST_TIMEOUT``, because that is a deliberate choice rather than
        an inherited default.

        Args:
            config: Configuration dictionary containing API settings

        Raises:
            ConfigurationError: If required configuration is missing, or if
                ``LOG_EXTRACT_TIMEOUT`` cannot be read as a positive number of
                seconds
        """
        super().__init__(config)

        # Validate required configuration
        validate_config_for_api(config, ["LOG_TOKEN"])

        self.auth_token = config["LOG_TOKEN"]

        # BaseAPIClient set self.timeout from REQUEST_TIMEOUT, whose 30s default
        # is tuned for the sub-second admin/browsing calls that dominate this
        # SDK. Extracts routinely run for minutes, so they get their own budget
        # pinned to the server-side lock TTL (see EXTRACT_LOCK_TTL).
        # LOG_EXTRACT_TIMEOUT overrides it for extraction only.
        inherited = self.timeout

        # "Unset" has to mean absent *or blank*, not just absent.
        # config/default.json always carries a LOG_EXTRACT_TIMEOUT key -- it is
        # a ${A8_LOG_EXTRACT_TIMEOUT:-} reference that resolves to an empty
        # string when the env var is not set. A bare `not in config` test
        # therefore never fired for anyone loading the shipped config, which
        # made the clamp below reachable only from a hand-built dict: exactly
        # the shape the unit tests used, and exactly the shape no real caller
        # has. The REQUEST_TIMEOUT=1800 workaround it exists to protect was
        # being cut to 600 in production while the tests stayed green.
        raw_timeout: Any = config.get("LOG_EXTRACT_TIMEOUT")
        explicitly_set = raw_timeout is not None and str(raw_timeout).strip() != ""
        timeout: Any = raw_timeout if explicitly_set else self.EXTRACT_LOCK_TTL
        try:
            self.timeout = int(timeout)
        except (TypeError, ValueError) as exc:
            raise ConfigurationError(
                "LOG_EXTRACT_TIMEOUT must be an integer number of seconds, "
                f"got {timeout!r}"
            ) from exc
        if self.timeout <= 0:
            raise ConfigurationError(
                f"LOG_EXTRACT_TIMEOUT must be positive, got {self.timeout}"
            )

        # Raising the extraction default must never *lower* a timeout someone
        # already raised by hand. Anyone who set REQUEST_TIMEOUT to 1800 to
        # work around the old 30s limit would otherwise be cut to the lock TTL
        # on upgrade -- a regression dressed up as a fix. Only the default is
        # clamped upward; an explicit LOG_EXTRACT_TIMEOUT still wins outright,
        # including a smaller one, because that is a deliberate choice.
        if not explicitly_set and isinstance(inherited, (int, float)):
            self.timeout = max(self.timeout, int(inherited))

        self.logger.info(
            "Initialized LogExtractionAPI client (timeout=%ss)", self.timeout
        )

    @staticmethod
    def _days_ago_to_epoch(days_ago: int) -> int:
        """Convert a 'days ago' integer to an epoch timestamp.

        Args:
            days_ago: Number of days before today (0 = today)

        Returns:
            Unix epoch timestamp (seconds) for midnight UTC of the target date

        Raises:
            ValidationError: If days_ago is negative
        """
        if days_ago < 0:
            raise ValidationError("days_ago must be a non-negative integer")
        target = datetime.now(UTC) - timedelta(days=days_ago)
        # Use midnight UTC of the target date
        midnight = target.replace(hour=0, minute=0, second=0, microsecond=0)
        return int(midnight.timestamp())

    @api_tag(MethodType.API_COMMAND, api_command="extractlog")
    def extract_logs(
        self,
        org: str,
        start_seq: int,
        log_types: list[str],
        end_seq: int | None = None,
        limit: int | None = None,
        start_date: int | datetime | None = None,
        end_date: int | datetime | None = None,
        include_suborgs: bool | None = None,
    ) -> dict[str, Any]:
        """Extract logs for the specified organization.

        Retrieves logs from the Silo platform for the specified organization
        within the given sequence range and log types. This method supports
        pagination for handling large datasets.

        Args:
            org: Organization name to extract logs from (API org name, not vanity URL).
            start_seq: Log sequence number from which to start extraction (must be >= 0).
            log_types: List of log type strings from ``VALID_LOG_TYPES``
                (e.g., ``["AUTH", "SESSION", "ADMIN_AUDIT"]``). Must be a
                non-empty list. Sent as a comma-joined string in the wire
                format (``"type": "AUTH,SESSION,ADMIN_AUDIT"``).
            end_seq: Optional log sequence number at which to stop extraction
                (must be >= start_seq).
            limit: Optional number of logs to return (max 1000 per request).
                If extracting ENC logs, the limit is automatically capped at
                ``MAX_ENC_BATCH_SIZE`` (30).
            start_date: Optional date filter. If an int, interpreted as "days
                ago from today" and converted to an epoch timestamp. If a
                datetime, converted directly to epoch. Sent as ``start_time``
                (epoch integer) in the wire format.
            end_date: Optional date filter. Same format rules as start_date.
                Sent as ``end_time`` (epoch integer) in the wire format.
            include_suborgs: If True, include logs from sub-organizations.
                Omitted from the request when not provided.

        Note:
            Wire format parameter mappings:

            - Python ``log_types`` (list) → wire ``"type"``
              (comma-joined string)
            - Python ``start_date``/``end_date`` (int or datetime) →
              wire ``"start_time"``/``"end_time"`` (epoch int)
            - Log extraction uses ``"org"`` (not ``"org_name"``) in the wire format
            - Four log types use **spaces** on the wire but **underscores**
              in the SDK: ``BLOCKED_URL`` → ``BLOCKED URL``,
              ``CASE_MANAGER`` → ``CASE MANAGER``,
              ``LOCATION_CHANGE`` → ``LOCATION CHANGE``,
              ``POST_DATA`` → ``POST DATA``.  The conversion is automatic.

            Wire format structure::

                {
                    "command": "extractlog",
                    "org": "my_company",  # ← uses "org", not "org_name"
                    "start_seq": 1000,
                    "type": "AUTH,SESSION,BLOCKED URL",  # ← spaces on wire
                    "start_time": 1704067200,  # ← start_date → start_time (epoch)
                    "end_time": 1707753600,  # ← end_date → end_time (epoch)
                    "limit": 500
                }

        Returns:
            Dictionary containing extracted logs and metadata including:

            - ``logs``: List of log entries matching the criteria
            - ``is_more``: Boolean indicating if more logs are available
            - ``next_seq``: Next sequence number for pagination
            - ``total_count``: Total number of logs in the range
            - ``org``: Organization name
            - ``log_types``: Types of logs included

        Raises:
            LogExtractionAPIError: If the API request fails. A
                ``PermissionDenied: log.extract already in progress``
                error means another extraction with the same parameters
                is running. The backend holds a per-extract lock that
                auto-expires after **600 seconds** (10 minutes). A
                background cleanup task also sweeps stale locks every
                5 minutes. Retry after the lock expires.
            ValidationError: If parameters are invalid or log types are unsupported

        Example:
            >>> # Extract recent logs with comma-joined type list
            >>> result = api.extract_logs(
            ...     org="my_company",
            ...     start_seq=1000,
            ...     log_types=["AUTH", "SESSION", "ADMIN_AUDIT"],
            ...     limit=500
            ... )
            >>> print(f"Retrieved {len(result['logs'])} logs")
            >>> if result['is_more']:
            ...     print(f"More logs available starting at seq {result['next_seq']}")
            >>>
            >>> # Date-based filtering (start_date as int = days ago)
            >>> result = api.extract_logs(
            ...     org="my_company",
            ...     start_seq=0,
            ...     log_types=["AUTH"],
            ...     start_date=7,  # ← 7 days ago
            ...     limit=1000
            ... )
        """
        # Validate parameters
        if not org or not isinstance(org, str):
            raise ValidationError("Organization name must be a non-empty string")

        if not isinstance(start_seq, int) or start_seq < 0:
            raise ValidationError("start_seq must be a non-negative integer")

        if not isinstance(log_types, list) or not log_types:
            raise ValidationError("log_types must be a non-empty list")

        # Validate log types
        invalid_types = set(log_types) - self.VALID_LOG_TYPES
        if invalid_types:
            raise ValidationError(
                f"Invalid log types: {', '.join(sorted(invalid_types))}. "
                f"Valid types are: {', '.join(sorted(self.VALID_LOG_TYPES))}"
            )

        if end_seq is not None:
            if not isinstance(end_seq, int) or end_seq < start_seq:
                raise ValidationError("end_seq must be an integer >= start_seq")

        if limit is not None:
            if not isinstance(limit, int) or limit <= 0 or limit > 1000:
                raise ValidationError("limit must be a positive integer <= 1000")

        # Enforce MAX_ENC_BATCH_SIZE for ENC log type
        if "ENC" in log_types:
            if limit is None or limit > self.MAX_ENC_BATCH_SIZE:
                self.logger.info(
                    "ENC log type detected — enforcing MAX_ENC_BATCH_SIZE limit of %d",
                    self.MAX_ENC_BATCH_SIZE,
                )
                limit = self.MAX_ENC_BATCH_SIZE

        # Build extraction payload — convert SDK type names to wire format
        wire_types = [_to_wire_type(lt) for lt in log_types]
        extract_params: dict[str, Any] = {
            "command": "extractlog",
            "org": org,
            "start_seq": start_seq,
            "type": ",".join(wire_types),
        }

        # Add optional parameters
        if end_seq is not None:
            extract_params["end_seq"] = end_seq

        if limit is not None:
            extract_params["limit"] = limit

        # Add date-based filtering
        if start_date is not None:
            if isinstance(start_date, int):
                extract_params["start_time"] = self._days_ago_to_epoch(start_date)
            elif isinstance(start_date, datetime):
                extract_params["start_time"] = int(start_date.timestamp())
            else:
                raise ValidationError(
                    "start_date must be an int (days ago) or datetime object"
                )

        if end_date is not None:
            if isinstance(end_date, int):
                extract_params["end_time"] = self._days_ago_to_epoch(end_date)
            elif isinstance(end_date, datetime):
                extract_params["end_time"] = int(end_date.timestamp())
            else:
                raise ValidationError(
                    "end_date must be an int (days ago) or datetime object"
                )

        if include_suborgs is not None:
            extract_params["include_suborgs"] = include_suborgs

        payload = [extract_params]

        try:
            self.logger.debug(
                "Extracting logs for org: %s, start_seq: %d, types: %s",
                org,
                start_seq,
                log_types,
            )

            response = self._make_api_request("POST", "api/", self.auth_token, payload)
            if not isinstance(response, list) or len(response) < 2:
                self.logger.warning("Unexpected response format: %s", response)
                raise LogExtractionAPIError(f"Unexpected response format: {response}")

            result = response[1]
            if isinstance(result, dict) and "error" in result:
                raise LogExtractionAPIError(f"API error: {result['error']}")
            # Unwrap standard {"result": {...}} envelope if present
            if isinstance(result, dict) and "result" in result and "logs" not in result:
                result = result["result"]

            if isinstance(result, dict):
                if "logs" not in result:
                    result["logs"] = []

                log_count = len(result.get("logs", []))
                self.logger.info(
                    "Successfully extracted %d logs for organization: %s",
                    log_count,
                    org,
                )
                return result

            raise LogExtractionAPIError(f"Unexpected result format: {result}")

        except Exception as e:
            if isinstance(e, LogExtractionAPIError):
                raise
            raise LogExtractionAPIError(f"Failed to extract logs: {e}") from e

    @api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
    def extract_all_logs(
        self,
        org: str,
        start_seq: int,
        log_types: list[str],
        end_seq: int | None = None,
        batch_size: int = 1000,
        max_logs: int | None = None,
        start_date: int | datetime | None = None,
        end_date: int | datetime | None = None,
    ) -> list[dict[str, Any]]:
        """Extract all logs for the specified organization with automatic pagination.

        Retrieves all available logs matching the criteria by automatically
        handling pagination. This method is useful for comprehensive log
        extraction and analysis.

        Args:
            org: Organization name to extract logs from
            start_seq: Log sequence number from which to start extraction
            log_types: List of log types to collect
            end_seq: Optional log sequence number at which to stop extraction
            batch_size: Number of logs to request in each API call (max 1000)
            max_logs: Optional maximum total number of logs to retrieve
            start_date: Optional date filter (int days-ago or datetime)
            end_date: Optional date filter (int days-ago or datetime)

        Returns:
            List of all log entries matching the criteria

        Raises:
            LogExtractionAPIError: If the API request fails
            ValidationError: If parameters are invalid

        Example:
            >>> all_logs = api.extract_all_logs(
            ...     org="my_company",
            ...     start_seq=1,
            ...     log_types=["AUTH", "ADMIN_AUDIT"],
            ...     batch_size=500,
            ...     max_logs=10000
            ... )
            >>> print(f"Retrieved {len(all_logs)} total logs")
        """
        # Validate batch_size
        if not isinstance(batch_size, int) or batch_size <= 0 or batch_size > 1000:
            raise ValidationError("batch_size must be a positive integer <= 1000")

        if max_logs is not None:
            if not isinstance(max_logs, int) or max_logs <= 0:
                raise ValidationError("max_logs must be a positive integer")

        all_logs: list[dict[str, Any]] = []
        current_seq = start_seq
        total_retrieved = 0

        self.logger.info(
            "Starting bulk log extraction for org: %s, types: %s", org, log_types
        )

        try:
            while True:
                # Calculate limit for this batch
                current_limit = batch_size
                if max_logs is not None:
                    remaining = max_logs - total_retrieved
                    if remaining <= 0:
                        break
                    current_limit = min(batch_size, remaining)

                # Extract batch of logs
                batch_result = self.extract_logs(
                    org=org,
                    start_seq=current_seq,
                    log_types=log_types,
                    end_seq=end_seq,
                    limit=current_limit,
                    start_date=start_date,
                    end_date=end_date,
                )

                # Add logs to collection
                batch_logs = batch_result.get("logs", [])
                if batch_logs:
                    all_logs.extend(batch_logs)
                    total_retrieved += len(batch_logs)

                    self.logger.debug(
                        "Retrieved batch of %d logs (total: %d)",
                        len(batch_logs),
                        total_retrieved,
                    )

                # Check if more logs are available
                if not batch_result.get("is_more", False):
                    self.logger.info("Reached end of available logs")
                    break

                # Get next sequence number
                next_seq = batch_result.get("next_seq")
                if next_seq is None:
                    raise LogExtractionAPIError(
                        "Failed to get next sequence number for pagination"
                    )

                current_seq = next_seq

                # Check if we've reached the end sequence
                if end_seq is not None and current_seq > end_seq:
                    self.logger.info("Reached specified end sequence: %d", end_seq)
                    break

            self.logger.info(
                "Completed bulk log extraction: %d total logs retrieved", len(all_logs)
            )
            return all_logs

        except Exception as e:
            self.logger.error(
                "Bulk log extraction failed after retrieving %d logs: %s",
                len(all_logs),
                e,
            )
            if isinstance(e, LogExtractionAPIError):
                raise
            raise LogExtractionAPIError(
                f"Failed during bulk log extraction: {e}"
            ) from e

    @api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
    def get_log_sequence_info(self, org: str) -> dict[str, Any]:
        """Get information about available log sequences for an organization.

        Probes the log sequence range by fetching the earliest available log
        entry via ``extractlog``. The ext API does not expose a dedicated
        sequence-info command, so this method derives what it can from a
        minimal extraction call.

        Note:
            ``max_seq`` and ``total_logs`` are **not available** from the ext
            API and are always returned as ``None``. Use ``is_more`` to check
            whether logs exist beyond the first entry, and ``next_seq`` to
            begin paginating from the earliest available log.

        Args:
            org: Organization name (API org name, not vanity URL).

        Returns:
            Dictionary containing:

            - ``org`` (str): Organization name
            - ``min_seq`` (int): Sequence ID of the earliest available log,
                or ``0`` if no logs exist
            - ``is_more`` (bool): ``True`` if logs exist beyond the first
                retrieved entry
            - ``next_seq`` (int): Sequence number to use as ``start_seq``
                for the next extraction page
            - ``max_seq`` (None): Not available via ext API
            - ``total_logs`` (None): Not available via ext API

        Raises:
            LogExtractionAPIError: If the probe extraction fails
            ValidationError: If org is invalid

        Example:
            >>> seq_info = api.get_log_sequence_info("my_company")
            >>> print(f"Earliest log at seq: {seq_info['min_seq']}")
            >>> print(f"More logs available: {seq_info['is_more']}")
            >>> # max_seq and total_logs are None — not available via ext API
        """
        if not org or not isinstance(org, str):
            raise ValidationError("Organization name must be a non-empty string")

        try:
            self.logger.debug("Probing log sequence info for org: %s", org)
            result = self.extract_logs(
                org=org,
                start_seq=0,
                log_types=["AUTH"],
                limit=1,
            )
            logs = result.get("logs", [])
            min_seq = logs[0]["seq_id"] if logs else 0
            self.logger.debug("Log sequence probe complete: min_seq=%s", min_seq)
            return {
                "org": org,
                "min_seq": min_seq,
                "is_more": result.get("is_more", False),
                "next_seq": result.get("next_seq", min_seq),
                "max_seq": None,
                "total_logs": None,
            }
        except Exception as e:
            if isinstance(e, LogExtractionAPIError):
                raise
            raise LogExtractionAPIError(f"Failed to get log sequence info: {e}") from e

    @api_tag(MethodType.UTILITY)
    @classmethod
    def get_valid_log_types(cls) -> list[str]:
        """Get a list of all valid log types supported by the platform.

        Returns:
            Sorted list of valid log type strings

        Example:
            >>> valid_types = LogExtractionAPI.get_valid_log_types()
            >>> print("Supported log types:")
            >>> for log_type in valid_types:
            ...     print(f"  - {log_type}")
        """
        return sorted(cls.VALID_LOG_TYPES)

    @api_tag(MethodType.UTILITY)
    @classmethod
    def validate_log_types(cls, log_types: Any) -> bool:
        """Validate that all provided log types are supported.

        Args:
            log_types: List of log type strings to validate

        Returns:
            True if all log types are valid, False otherwise

        Example:
            >>> types_to_check = ["AUTH", "SESSION", "INVALID_TYPE"]
            >>> if not LogExtractionAPI.validate_log_types(types_to_check):
            ...     print("Some log types are invalid")
        """
        if not isinstance(log_types, list):
            return False

        return set(log_types).issubset(cls.VALID_LOG_TYPES)

    @api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
    def export_logs_to_file(
        self,
        org: str,
        start_seq: int,
        log_types: list[str],
        output_file: str,
        format_type: str = "json",
        end_seq: int | None = None,
        batch_size: int = 1000,
        start_date: int | datetime | None = None,
        end_date: int | datetime | None = None,
    ) -> dict[str, Any]:
        """Export logs to a file in the specified format.

        Extracts all matching logs and exports them to a file for external
        analysis, compliance reporting, or archival purposes.

        Note:
            CSV export escapes formula-injection risk: any field value
            starting with ``=``, ``+``, ``-``, ``@``, tab, or carriage
            return (e.g. a URL or clipboard value crafted as
            ``=HYPERLINK(...)``) is written with a leading single quote
            so spreadsheet applications (Excel, Google Sheets) treat it
            as literal text instead of evaluating it as a formula. JSON
            and text export are unaffected.

        Args:
            org: Organization name to extract logs from
            start_seq: Log sequence number from which to start extraction
            log_types: List of log types to collect
            output_file: Path to the output file
            format_type: Export format ("json", "csv", or "txt")
            end_seq: Optional log sequence number at which to stop extraction
            batch_size: Number of logs to process in each batch
            start_date: Optional date filter (int days-ago or datetime)
            end_date: Optional date filter (int days-ago or datetime)

        Returns:
            Dictionary containing export statistics including:
            - total_logs: Number of logs exported
            - output_file: Path to the exported file
            - format: Export format used
            - file_size: Size of the exported file in bytes

        Raises:
            LogExtractionAPIError: If export fails
            ValidationError: If parameters are invalid

        Example:
            >>> export_result = api.export_logs_to_file(
            ...     org="my_company",
            ...     start_seq=1,
            ...     log_types=["AUTH", "ADMIN_AUDIT"],
            ...     output_file="/path/to/audit_logs.json",
            ...     format_type="json"
            ... )
            >>> print(f"Exported {export_result['total_logs']} logs")
        """
        import csv
        import json
        from pathlib import Path

        # Validate format
        if format_type not in ["json", "csv", "txt"]:
            raise ValidationError("format_type must be 'json', 'csv', or 'txt'")

        if not output_file:
            raise ValidationError("output_file must be provided")

        # Extract all logs
        all_logs = self.extract_all_logs(
            org=org,
            start_seq=start_seq,
            log_types=log_types,
            end_seq=end_seq,
            batch_size=batch_size,
            start_date=start_date,
            end_date=end_date,
        )

        output_path = Path(output_file)
        output_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            if format_type == "json":
                with open(output_path, "w", encoding="utf-8") as f:
                    json.dump(all_logs, f, indent=2, default=str)

            elif format_type == "csv":
                if all_logs:
                    from silo_sdk.logging.log_schemas import get_fieldnames

                    fieldnames = get_fieldnames(log_types, all_logs)
                    sanitized_logs = [_sanitize_csv_row(log) for log in all_logs]

                    with open(output_path, "w", newline="", encoding="utf-8") as f:
                        writer = csv.DictWriter(
                            f, fieldnames=fieldnames, extrasaction="ignore"
                        )
                        writer.writeheader()
                        writer.writerows(sanitized_logs)

            elif format_type == "txt":
                with open(output_path, "w", encoding="utf-8") as f:
                    for i, log in enumerate(all_logs):
                        f.write(f"Log {i + 1}:\n")
                        for key, value in log.items():
                            f.write(f"  {key}: {value}\n")
                        f.write("\n")

            # Get file size
            file_size = output_path.stat().st_size

            self.logger.info(
                "Successfully exported %d logs to %s (%d bytes)",
                len(all_logs),
                output_file,
                file_size,
            )

            return {
                "total_logs": len(all_logs),
                "output_file": str(output_path),
                "format": format_type,
                "file_size": file_size,
            }

        except Exception as e:
            raise LogExtractionAPIError(f"Failed to export logs to file: {e}") from e

    @api_tag(MethodType.UTILITY)
    def group_logs_by_type(
        self, logs: list[dict[str, Any]]
    ) -> dict[str, list[dict[str, Any]]]:
        """Group a flat list of log entries by their type field.

        Args:
            logs: List of log entry dictionaries

        Returns:
            Dictionary mapping log type name to list of entries for that type.
            Entries without a type field are grouped under "UNKNOWN".

        Example:
            >>> logs = [
            ...     {"type": "AUTH", "user": "alice"},
            ...     {"type": "SESSION", "session_id": "s1"},
            ...     {"type": "AUTH", "user": "bob"},
            ...     {"no_type_field": "value"}
            ... ]
            >>> grouped = api.group_logs_by_type(logs)
            >>> len(grouped["AUTH"])
            2
            >>> len(grouped["UNKNOWN"])
            1
        """
        grouped: dict[str, list[dict[str, Any]]] = {}

        for entry in logs:
            log_type = entry.get("type", "UNKNOWN")
            if log_type not in grouped:
                grouped[log_type] = []
            grouped[log_type].append(entry)

        return grouped

__init__

__init__(config: dict[str, Any]) -> None

Initialize the log extraction API client.

This client takes its request timeout from LOG_EXTRACT_TIMEOUT (A8_LOG_EXTRACT_TIMEOUT), which defaults to 600 seconds -- the server's per-extract lock TTL -- rather than from the SDK-wide REQUEST_TIMEOUT default of 30 seconds. Extracts routinely run for minutes, and a client that gives up while the lock is still held blocks its own retries until that TTL expires.

The 600-second default only ever raises the extraction timeout; it never lowers one already set higher. A REQUEST_TIMEOUT above 600 is kept as-is, so anyone who raised it to work around the previous 30-second limit is not cut back on upgrade. An explicit LOG_EXTRACT_TIMEOUT wins outright, including a value below REQUEST_TIMEOUT, because that is a deliberate choice rather than an inherited default.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary containing API settings	required

Raises:

Type	Description
ConfigurationError	If required configuration is missing, or if LOG_EXTRACT_TIMEOUT cannot be read as a positive number of seconds

Source code in silo_sdk/logging/extraction_api.py

def __init__(self, config: dict[str, Any]) -> None:
    """Initialize the log extraction API client.

    This client takes its request timeout from ``LOG_EXTRACT_TIMEOUT``
    (``A8_LOG_EXTRACT_TIMEOUT``), which defaults to 600 seconds -- the
    server's per-extract lock TTL -- rather than from the SDK-wide
    ``REQUEST_TIMEOUT`` default of 30 seconds. Extracts routinely run for
    minutes, and a client that gives up while the lock is still held
    blocks its own retries until that TTL expires.

    The 600-second default only ever raises the extraction timeout; it
    never lowers one already set higher. A ``REQUEST_TIMEOUT`` above 600
    is kept as-is, so anyone who raised it to work around the previous
    30-second limit is not cut back on upgrade. An explicit
    ``LOG_EXTRACT_TIMEOUT`` wins outright, including a value below
    ``REQUEST_TIMEOUT``, because that is a deliberate choice rather than
    an inherited default.

    Args:
        config: Configuration dictionary containing API settings

    Raises:
        ConfigurationError: If required configuration is missing, or if
            ``LOG_EXTRACT_TIMEOUT`` cannot be read as a positive number of
            seconds
    """
    super().__init__(config)

    # Validate required configuration
    validate_config_for_api(config, ["LOG_TOKEN"])

    self.auth_token = config["LOG_TOKEN"]

    # BaseAPIClient set self.timeout from REQUEST_TIMEOUT, whose 30s default
    # is tuned for the sub-second admin/browsing calls that dominate this
    # SDK. Extracts routinely run for minutes, so they get their own budget
    # pinned to the server-side lock TTL (see EXTRACT_LOCK_TTL).
    # LOG_EXTRACT_TIMEOUT overrides it for extraction only.
    inherited = self.timeout

    # "Unset" has to mean absent *or blank*, not just absent.
    # config/default.json always carries a LOG_EXTRACT_TIMEOUT key -- it is
    # a ${A8_LOG_EXTRACT_TIMEOUT:-} reference that resolves to an empty
    # string when the env var is not set. A bare `not in config` test
    # therefore never fired for anyone loading the shipped config, which
    # made the clamp below reachable only from a hand-built dict: exactly
    # the shape the unit tests used, and exactly the shape no real caller
    # has. The REQUEST_TIMEOUT=1800 workaround it exists to protect was
    # being cut to 600 in production while the tests stayed green.
    raw_timeout: Any = config.get("LOG_EXTRACT_TIMEOUT")
    explicitly_set = raw_timeout is not None and str(raw_timeout).strip() != ""
    timeout: Any = raw_timeout if explicitly_set else self.EXTRACT_LOCK_TTL
    try:
        self.timeout = int(timeout)
    except (TypeError, ValueError) as exc:
        raise ConfigurationError(
            "LOG_EXTRACT_TIMEOUT must be an integer number of seconds, "
            f"got {timeout!r}"
        ) from exc
    if self.timeout <= 0:
        raise ConfigurationError(
            f"LOG_EXTRACT_TIMEOUT must be positive, got {self.timeout}"
        )

    # Raising the extraction default must never *lower* a timeout someone
    # already raised by hand. Anyone who set REQUEST_TIMEOUT to 1800 to
    # work around the old 30s limit would otherwise be cut to the lock TTL
    # on upgrade -- a regression dressed up as a fix. Only the default is
    # clamped upward; an explicit LOG_EXTRACT_TIMEOUT still wins outright,
    # including a smaller one, because that is a deliberate choice.
    if not explicitly_set and isinstance(inherited, (int, float)):
        self.timeout = max(self.timeout, int(inherited))

    self.logger.info(
        "Initialized LogExtractionAPI client (timeout=%ss)", self.timeout
    )

extract_logs

extract_logs(org: str, start_seq: int, log_types: list[str], end_seq: int | None = None, limit: int | None = None, start_date: int | datetime | None = None, end_date: int | datetime | None = None, include_suborgs: bool | None = None) -> dict[str, Any]

Extract logs for the specified organization.

Retrieves logs from the Silo platform for the specified organization within the given sequence range and log types. This method supports pagination for handling large datasets.

Parameters:

Name	Type	Description	Default
org	str	Organization name to extract logs from (API org name, not vanity URL).	required
start_seq	int	Log sequence number from which to start extraction (must be >= 0).	required
log_types	list[str]	List of log type strings from VALID_LOG_TYPES (e.g., ["AUTH", "SESSION", "ADMIN_AUDIT"]). Must be a non-empty list. Sent as a comma-joined string in the wire format ("type": "AUTH,SESSION,ADMIN_AUDIT").	required
end_seq	int | None	Optional log sequence number at which to stop extraction (must be >= start_seq).	None
limit	int | None	Optional number of logs to return (max 1000 per request). If extracting ENC logs, the limit is automatically capped at MAX_ENC_BATCH_SIZE (30).	None
start_date	int | datetime | None	Optional date filter. If an int, interpreted as "days ago from today" and converted to an epoch timestamp. If a datetime, converted directly to epoch. Sent as start_time (epoch integer) in the wire format.	None
end_date	int | datetime | None	Optional date filter. Same format rules as start_date. Sent as end_time (epoch integer) in the wire format.	None
include_suborgs	bool | None	If True, include logs from sub-organizations. Omitted from the request when not provided.	None

Note

Wire format parameter mappings:

• Python log_types (list) → wire "type" (comma-joined string)
• Python start_date/end_date (int or datetime) → wire "start_time"/"end_time" (epoch int)
• Log extraction uses "org" (not "org_name") in the wire format
• Four log types use spaces on the wire but underscores in the SDK: BLOCKED_URL → BLOCKED URL, CASE_MANAGER → CASE MANAGER, LOCATION_CHANGE → LOCATION CHANGE, POST_DATA → POST DATA. The conversion is automatic.

Wire format structure::

{
    "command": "extractlog",
    "org": "my_company",  # ← uses "org", not "org_name"
    "start_seq": 1000,
    "type": "AUTH,SESSION,BLOCKED URL",  # ← spaces on wire
    "start_time": 1704067200,  # ← start_date → start_time (epoch)
    "end_time": 1707753600,  # ← end_date → end_time (epoch)
    "limit": 500
}

Returns:

Type	Description
dict[str, Any]	Dictionary containing extracted logs and metadata including:
dict[str, Any]	logs: List of log entries matching the criteria
dict[str, Any]	is_more: Boolean indicating if more logs are available
dict[str, Any]	next_seq: Next sequence number for pagination
dict[str, Any]	total_count: Total number of logs in the range
dict[str, Any]	org: Organization name
dict[str, Any]	log_types: Types of logs included

Raises:

Type	Description
LogExtractionAPIError	If the API request fails. A PermissionDenied: log.extract already in progress error means another extraction with the same parameters is running. The backend holds a per-extract lock that auto-expires after 600 seconds (10 minutes). A background cleanup task also sweeps stale locks every 5 minutes. Retry after the lock expires.
ValidationError	If parameters are invalid or log types are unsupported

Example

Extract recent logs with comma-joined type list

result = api.extract_logs( ... org="my_company", ... start_seq=1000, ... log_types=["AUTH", "SESSION", "ADMIN_AUDIT"], ... limit=500 ... ) print(f"Retrieved {len(result['logs'])} logs") if result['is_more']: ... print(f"More logs available starting at seq {result['next_seq']}")

Date-based filtering (start_date as int = days ago)

result = api.extract_logs( ... org="my_company", ... start_seq=0, ... log_types=["AUTH"], ... start_date=7, # ← 7 days ago ... limit=1000 ... )

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.API_COMMAND, api_command="extractlog")
def extract_logs(
    self,
    org: str,
    start_seq: int,
    log_types: list[str],
    end_seq: int | None = None,
    limit: int | None = None,
    start_date: int | datetime | None = None,
    end_date: int | datetime | None = None,
    include_suborgs: bool | None = None,
) -> dict[str, Any]:
    """Extract logs for the specified organization.

    Retrieves logs from the Silo platform for the specified organization
    within the given sequence range and log types. This method supports
    pagination for handling large datasets.

    Args:
        org: Organization name to extract logs from (API org name, not vanity URL).
        start_seq: Log sequence number from which to start extraction (must be >= 0).
        log_types: List of log type strings from ``VALID_LOG_TYPES``
            (e.g., ``["AUTH", "SESSION", "ADMIN_AUDIT"]``). Must be a
            non-empty list. Sent as a comma-joined string in the wire
            format (``"type": "AUTH,SESSION,ADMIN_AUDIT"``).
        end_seq: Optional log sequence number at which to stop extraction
            (must be >= start_seq).
        limit: Optional number of logs to return (max 1000 per request).
            If extracting ENC logs, the limit is automatically capped at
            ``MAX_ENC_BATCH_SIZE`` (30).
        start_date: Optional date filter. If an int, interpreted as "days
            ago from today" and converted to an epoch timestamp. If a
            datetime, converted directly to epoch. Sent as ``start_time``
            (epoch integer) in the wire format.
        end_date: Optional date filter. Same format rules as start_date.
            Sent as ``end_time`` (epoch integer) in the wire format.
        include_suborgs: If True, include logs from sub-organizations.
            Omitted from the request when not provided.

    Note:
        Wire format parameter mappings:

        - Python ``log_types`` (list) → wire ``"type"``
          (comma-joined string)
        - Python ``start_date``/``end_date`` (int or datetime) →
          wire ``"start_time"``/``"end_time"`` (epoch int)
        - Log extraction uses ``"org"`` (not ``"org_name"``) in the wire format
        - Four log types use **spaces** on the wire but **underscores**
          in the SDK: ``BLOCKED_URL`` → ``BLOCKED URL``,
          ``CASE_MANAGER`` → ``CASE MANAGER``,
          ``LOCATION_CHANGE`` → ``LOCATION CHANGE``,
          ``POST_DATA`` → ``POST DATA``.  The conversion is automatic.

        Wire format structure::

            {
                "command": "extractlog",
                "org": "my_company",  # ← uses "org", not "org_name"
                "start_seq": 1000,
                "type": "AUTH,SESSION,BLOCKED URL",  # ← spaces on wire
                "start_time": 1704067200,  # ← start_date → start_time (epoch)
                "end_time": 1707753600,  # ← end_date → end_time (epoch)
                "limit": 500
            }

    Returns:
        Dictionary containing extracted logs and metadata including:

        - ``logs``: List of log entries matching the criteria
        - ``is_more``: Boolean indicating if more logs are available
        - ``next_seq``: Next sequence number for pagination
        - ``total_count``: Total number of logs in the range
        - ``org``: Organization name
        - ``log_types``: Types of logs included

    Raises:
        LogExtractionAPIError: If the API request fails. A
            ``PermissionDenied: log.extract already in progress``
            error means another extraction with the same parameters
            is running. The backend holds a per-extract lock that
            auto-expires after **600 seconds** (10 minutes). A
            background cleanup task also sweeps stale locks every
            5 minutes. Retry after the lock expires.
        ValidationError: If parameters are invalid or log types are unsupported

    Example:
        >>> # Extract recent logs with comma-joined type list
        >>> result = api.extract_logs(
        ...     org="my_company",
        ...     start_seq=1000,
        ...     log_types=["AUTH", "SESSION", "ADMIN_AUDIT"],
        ...     limit=500
        ... )
        >>> print(f"Retrieved {len(result['logs'])} logs")
        >>> if result['is_more']:
        ...     print(f"More logs available starting at seq {result['next_seq']}")
        >>>
        >>> # Date-based filtering (start_date as int = days ago)
        >>> result = api.extract_logs(
        ...     org="my_company",
        ...     start_seq=0,
        ...     log_types=["AUTH"],
        ...     start_date=7,  # ← 7 days ago
        ...     limit=1000
        ... )
    """
    # Validate parameters
    if not org or not isinstance(org, str):
        raise ValidationError("Organization name must be a non-empty string")

    if not isinstance(start_seq, int) or start_seq < 0:
        raise ValidationError("start_seq must be a non-negative integer")

    if not isinstance(log_types, list) or not log_types:
        raise ValidationError("log_types must be a non-empty list")

    # Validate log types
    invalid_types = set(log_types) - self.VALID_LOG_TYPES
    if invalid_types:
        raise ValidationError(
            f"Invalid log types: {', '.join(sorted(invalid_types))}. "
            f"Valid types are: {', '.join(sorted(self.VALID_LOG_TYPES))}"
        )

    if end_seq is not None:
        if not isinstance(end_seq, int) or end_seq < start_seq:
            raise ValidationError("end_seq must be an integer >= start_seq")

    if limit is not None:
        if not isinstance(limit, int) or limit <= 0 or limit > 1000:
            raise ValidationError("limit must be a positive integer <= 1000")

    # Enforce MAX_ENC_BATCH_SIZE for ENC log type
    if "ENC" in log_types:
        if limit is None or limit > self.MAX_ENC_BATCH_SIZE:
            self.logger.info(
                "ENC log type detected — enforcing MAX_ENC_BATCH_SIZE limit of %d",
                self.MAX_ENC_BATCH_SIZE,
            )
            limit = self.MAX_ENC_BATCH_SIZE

    # Build extraction payload — convert SDK type names to wire format
    wire_types = [_to_wire_type(lt) for lt in log_types]
    extract_params: dict[str, Any] = {
        "command": "extractlog",
        "org": org,
        "start_seq": start_seq,
        "type": ",".join(wire_types),
    }

    # Add optional parameters
    if end_seq is not None:
        extract_params["end_seq"] = end_seq

    if limit is not None:
        extract_params["limit"] = limit

    # Add date-based filtering
    if start_date is not None:
        if isinstance(start_date, int):
            extract_params["start_time"] = self._days_ago_to_epoch(start_date)
        elif isinstance(start_date, datetime):
            extract_params["start_time"] = int(start_date.timestamp())
        else:
            raise ValidationError(
                "start_date must be an int (days ago) or datetime object"
            )

    if end_date is not None:
        if isinstance(end_date, int):
            extract_params["end_time"] = self._days_ago_to_epoch(end_date)
        elif isinstance(end_date, datetime):
            extract_params["end_time"] = int(end_date.timestamp())
        else:
            raise ValidationError(
                "end_date must be an int (days ago) or datetime object"
            )

    if include_suborgs is not None:
        extract_params["include_suborgs"] = include_suborgs

    payload = [extract_params]

    try:
        self.logger.debug(
            "Extracting logs for org: %s, start_seq: %d, types: %s",
            org,
            start_seq,
            log_types,
        )

        response = self._make_api_request("POST", "api/", self.auth_token, payload)
        if not isinstance(response, list) or len(response) < 2:
            self.logger.warning("Unexpected response format: %s", response)
            raise LogExtractionAPIError(f"Unexpected response format: {response}")

        result = response[1]
        if isinstance(result, dict) and "error" in result:
            raise LogExtractionAPIError(f"API error: {result['error']}")
        # Unwrap standard {"result": {...}} envelope if present
        if isinstance(result, dict) and "result" in result and "logs" not in result:
            result = result["result"]

        if isinstance(result, dict):
            if "logs" not in result:
                result["logs"] = []

            log_count = len(result.get("logs", []))
            self.logger.info(
                "Successfully extracted %d logs for organization: %s",
                log_count,
                org,
            )
            return result

        raise LogExtractionAPIError(f"Unexpected result format: {result}")

    except Exception as e:
        if isinstance(e, LogExtractionAPIError):
            raise
        raise LogExtractionAPIError(f"Failed to extract logs: {e}") from e

extract_all_logs

extract_all_logs(org: str, start_seq: int, log_types: list[str], end_seq: int | None = None, batch_size: int = 1000, max_logs: int | None = None, start_date: int | datetime | None = None, end_date: int | datetime | None = None) -> list[dict[str, Any]]

Extract all logs for the specified organization with automatic pagination.

Retrieves all available logs matching the criteria by automatically handling pagination. This method is useful for comprehensive log extraction and analysis.

Parameters:

Name	Type	Description	Default
org	str	Organization name to extract logs from	required
start_seq	int	Log sequence number from which to start extraction	required
log_types	list[str]	List of log types to collect	required
end_seq	int | None	Optional log sequence number at which to stop extraction	None
batch_size	int	Number of logs to request in each API call (max 1000)	1000
max_logs	int | None	Optional maximum total number of logs to retrieve	None
start_date	int | datetime | None	Optional date filter (int days-ago or datetime)	None
end_date	int | datetime | None	Optional date filter (int days-ago or datetime)	None

Returns:

Type	Description
list[dict[str, Any]]	List of all log entries matching the criteria

Raises:

Type	Description
LogExtractionAPIError	If the API request fails
ValidationError	If parameters are invalid

Example

all_logs = api.extract_all_logs( ... org="my_company", ... start_seq=1, ... log_types=["AUTH", "ADMIN_AUDIT"], ... batch_size=500, ... max_logs=10000 ... ) print(f"Retrieved {len(all_logs)} total logs")

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
def extract_all_logs(
    self,
    org: str,
    start_seq: int,
    log_types: list[str],
    end_seq: int | None = None,
    batch_size: int = 1000,
    max_logs: int | None = None,
    start_date: int | datetime | None = None,
    end_date: int | datetime | None = None,
) -> list[dict[str, Any]]:
    """Extract all logs for the specified organization with automatic pagination.

    Retrieves all available logs matching the criteria by automatically
    handling pagination. This method is useful for comprehensive log
    extraction and analysis.

    Args:
        org: Organization name to extract logs from
        start_seq: Log sequence number from which to start extraction
        log_types: List of log types to collect
        end_seq: Optional log sequence number at which to stop extraction
        batch_size: Number of logs to request in each API call (max 1000)
        max_logs: Optional maximum total number of logs to retrieve
        start_date: Optional date filter (int days-ago or datetime)
        end_date: Optional date filter (int days-ago or datetime)

    Returns:
        List of all log entries matching the criteria

    Raises:
        LogExtractionAPIError: If the API request fails
        ValidationError: If parameters are invalid

    Example:
        >>> all_logs = api.extract_all_logs(
        ...     org="my_company",
        ...     start_seq=1,
        ...     log_types=["AUTH", "ADMIN_AUDIT"],
        ...     batch_size=500,
        ...     max_logs=10000
        ... )
        >>> print(f"Retrieved {len(all_logs)} total logs")
    """
    # Validate batch_size
    if not isinstance(batch_size, int) or batch_size <= 0 or batch_size > 1000:
        raise ValidationError("batch_size must be a positive integer <= 1000")

    if max_logs is not None:
        if not isinstance(max_logs, int) or max_logs <= 0:
            raise ValidationError("max_logs must be a positive integer")

    all_logs: list[dict[str, Any]] = []
    current_seq = start_seq
    total_retrieved = 0

    self.logger.info(
        "Starting bulk log extraction for org: %s, types: %s", org, log_types
    )

    try:
        while True:
            # Calculate limit for this batch
            current_limit = batch_size
            if max_logs is not None:
                remaining = max_logs - total_retrieved
                if remaining <= 0:
                    break
                current_limit = min(batch_size, remaining)

            # Extract batch of logs
            batch_result = self.extract_logs(
                org=org,
                start_seq=current_seq,
                log_types=log_types,
                end_seq=end_seq,
                limit=current_limit,
                start_date=start_date,
                end_date=end_date,
            )

            # Add logs to collection
            batch_logs = batch_result.get("logs", [])
            if batch_logs:
                all_logs.extend(batch_logs)
                total_retrieved += len(batch_logs)

                self.logger.debug(
                    "Retrieved batch of %d logs (total: %d)",
                    len(batch_logs),
                    total_retrieved,
                )

            # Check if more logs are available
            if not batch_result.get("is_more", False):
                self.logger.info("Reached end of available logs")
                break

            # Get next sequence number
            next_seq = batch_result.get("next_seq")
            if next_seq is None:
                raise LogExtractionAPIError(
                    "Failed to get next sequence number for pagination"
                )

            current_seq = next_seq

            # Check if we've reached the end sequence
            if end_seq is not None and current_seq > end_seq:
                self.logger.info("Reached specified end sequence: %d", end_seq)
                break

        self.logger.info(
            "Completed bulk log extraction: %d total logs retrieved", len(all_logs)
        )
        return all_logs

    except Exception as e:
        self.logger.error(
            "Bulk log extraction failed after retrieving %d logs: %s",
            len(all_logs),
            e,
        )
        if isinstance(e, LogExtractionAPIError):
            raise
        raise LogExtractionAPIError(
            f"Failed during bulk log extraction: {e}"
        ) from e

get_log_sequence_info

get_log_sequence_info(org: str) -> dict[str, Any]

Get information about available log sequences for an organization.

Probes the log sequence range by fetching the earliest available log entry via extractlog. The ext API does not expose a dedicated sequence-info command, so this method derives what it can from a minimal extraction call.

Note

max_seq and total_logs are not available from the ext API and are always returned as None. Use is_more to check whether logs exist beyond the first entry, and next_seq to begin paginating from the earliest available log.

Parameters:

Name	Type	Description	Default
org	str	Organization name (API org name, not vanity URL).	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing:
dict[str, Any]	org (str): Organization name
dict[str, Any]	min_seq (int): Sequence ID of the earliest available log, or 0 if no logs exist
dict[str, Any]	is_more (bool): True if logs exist beyond the first retrieved entry
dict[str, Any]	next_seq (int): Sequence number to use as start_seq for the next extraction page
dict[str, Any]	max_seq (None): Not available via ext API
dict[str, Any]	total_logs (None): Not available via ext API

Raises:

Type	Description
LogExtractionAPIError	If the probe extraction fails
ValidationError	If org is invalid

Example

seq_info = api.get_log_sequence_info("my_company") print(f"Earliest log at seq: {seq_info['min_seq']}") print(f"More logs available: {seq_info['is_more']}")

max_seq and total_logs are None — not available via ext API

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
def get_log_sequence_info(self, org: str) -> dict[str, Any]:
    """Get information about available log sequences for an organization.

    Probes the log sequence range by fetching the earliest available log
    entry via ``extractlog``. The ext API does not expose a dedicated
    sequence-info command, so this method derives what it can from a
    minimal extraction call.

    Note:
        ``max_seq`` and ``total_logs`` are **not available** from the ext
        API and are always returned as ``None``. Use ``is_more`` to check
        whether logs exist beyond the first entry, and ``next_seq`` to
        begin paginating from the earliest available log.

    Args:
        org: Organization name (API org name, not vanity URL).

    Returns:
        Dictionary containing:

        - ``org`` (str): Organization name
        - ``min_seq`` (int): Sequence ID of the earliest available log,
            or ``0`` if no logs exist
        - ``is_more`` (bool): ``True`` if logs exist beyond the first
            retrieved entry
        - ``next_seq`` (int): Sequence number to use as ``start_seq``
            for the next extraction page
        - ``max_seq`` (None): Not available via ext API
        - ``total_logs`` (None): Not available via ext API

    Raises:
        LogExtractionAPIError: If the probe extraction fails
        ValidationError: If org is invalid

    Example:
        >>> seq_info = api.get_log_sequence_info("my_company")
        >>> print(f"Earliest log at seq: {seq_info['min_seq']}")
        >>> print(f"More logs available: {seq_info['is_more']}")
        >>> # max_seq and total_logs are None — not available via ext API
    """
    if not org or not isinstance(org, str):
        raise ValidationError("Organization name must be a non-empty string")

    try:
        self.logger.debug("Probing log sequence info for org: %s", org)
        result = self.extract_logs(
            org=org,
            start_seq=0,
            log_types=["AUTH"],
            limit=1,
        )
        logs = result.get("logs", [])
        min_seq = logs[0]["seq_id"] if logs else 0
        self.logger.debug("Log sequence probe complete: min_seq=%s", min_seq)
        return {
            "org": org,
            "min_seq": min_seq,
            "is_more": result.get("is_more", False),
            "next_seq": result.get("next_seq", min_seq),
            "max_seq": None,
            "total_logs": None,
        }
    except Exception as e:
        if isinstance(e, LogExtractionAPIError):
            raise
        raise LogExtractionAPIError(f"Failed to get log sequence info: {e}") from e

get_valid_log_types classmethod

get_valid_log_types() -> list[str]

Get a list of all valid log types supported by the platform.

Returns:

Type	Description
list[str]	Sorted list of valid log type strings

Example

valid_types = LogExtractionAPI.get_valid_log_types() print("Supported log types:") for log_type in valid_types: ... print(f" - {log_type}")

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.UTILITY)
@classmethod
def get_valid_log_types(cls) -> list[str]:
    """Get a list of all valid log types supported by the platform.

    Returns:
        Sorted list of valid log type strings

    Example:
        >>> valid_types = LogExtractionAPI.get_valid_log_types()
        >>> print("Supported log types:")
        >>> for log_type in valid_types:
        ...     print(f"  - {log_type}")
    """
    return sorted(cls.VALID_LOG_TYPES)

validate_log_types classmethod

validate_log_types(log_types: Any) -> bool

Validate that all provided log types are supported.

Parameters:

Name	Type	Description	Default
log_types	Any	List of log type strings to validate	required

Returns:

Type	Description
bool	True if all log types are valid, False otherwise

Example

types_to_check = ["AUTH", "SESSION", "INVALID_TYPE"] if not LogExtractionAPI.validate_log_types(types_to_check): ... print("Some log types are invalid")

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.UTILITY)
@classmethod
def validate_log_types(cls, log_types: Any) -> bool:
    """Validate that all provided log types are supported.

    Args:
        log_types: List of log type strings to validate

    Returns:
        True if all log types are valid, False otherwise

    Example:
        >>> types_to_check = ["AUTH", "SESSION", "INVALID_TYPE"]
        >>> if not LogExtractionAPI.validate_log_types(types_to_check):
        ...     print("Some log types are invalid")
    """
    if not isinstance(log_types, list):
        return False

    return set(log_types).issubset(cls.VALID_LOG_TYPES)

export_logs_to_file

export_logs_to_file(org: str, start_seq: int, log_types: list[str], output_file: str, format_type: str = 'json', end_seq: int | None = None, batch_size: int = 1000, start_date: int | datetime | None = None, end_date: int | datetime | None = None) -> dict[str, Any]

Export logs to a file in the specified format.

Extracts all matching logs and exports them to a file for external analysis, compliance reporting, or archival purposes.

Note

CSV export escapes formula-injection risk: any field value starting with =, +, -, @, tab, or carriage return (e.g. a URL or clipboard value crafted as =HYPERLINK(...)) is written with a leading single quote so spreadsheet applications (Excel, Google Sheets) treat it as literal text instead of evaluating it as a formula. JSON and text export are unaffected.

Parameters:

Name	Type	Description	Default
org	str	Organization name to extract logs from	required
start_seq	int	Log sequence number from which to start extraction	required
log_types	list[str]	List of log types to collect	required
output_file	str	Path to the output file	required
format_type	str	Export format ("json", "csv", or "txt")	'json'
end_seq	int | None	Optional log sequence number at which to stop extraction	None
batch_size	int	Number of logs to process in each batch	1000
start_date	int | datetime | None	Optional date filter (int days-ago or datetime)	None
end_date	int | datetime | None	Optional date filter (int days-ago or datetime)	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing export statistics including:
dict[str, Any]	total_logs: Number of logs exported
dict[str, Any]	output_file: Path to the exported file
dict[str, Any]	format: Export format used
dict[str, Any]	file_size: Size of the exported file in bytes

Raises:

Type	Description
LogExtractionAPIError	If export fails
ValidationError	If parameters are invalid

Example

export_result = api.export_logs_to_file( ... org="my_company", ... start_seq=1, ... log_types=["AUTH", "ADMIN_AUDIT"], ... output_file="/path/to/audit_logs.json", ... format_type="json" ... ) print(f"Exported {export_result['total_logs']} logs")

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["extractlog"])
def export_logs_to_file(
    self,
    org: str,
    start_seq: int,
    log_types: list[str],
    output_file: str,
    format_type: str = "json",
    end_seq: int | None = None,
    batch_size: int = 1000,
    start_date: int | datetime | None = None,
    end_date: int | datetime | None = None,
) -> dict[str, Any]:
    """Export logs to a file in the specified format.

    Extracts all matching logs and exports them to a file for external
    analysis, compliance reporting, or archival purposes.

    Note:
        CSV export escapes formula-injection risk: any field value
        starting with ``=``, ``+``, ``-``, ``@``, tab, or carriage
        return (e.g. a URL or clipboard value crafted as
        ``=HYPERLINK(...)``) is written with a leading single quote
        so spreadsheet applications (Excel, Google Sheets) treat it
        as literal text instead of evaluating it as a formula. JSON
        and text export are unaffected.

    Args:
        org: Organization name to extract logs from
        start_seq: Log sequence number from which to start extraction
        log_types: List of log types to collect
        output_file: Path to the output file
        format_type: Export format ("json", "csv", or "txt")
        end_seq: Optional log sequence number at which to stop extraction
        batch_size: Number of logs to process in each batch
        start_date: Optional date filter (int days-ago or datetime)
        end_date: Optional date filter (int days-ago or datetime)

    Returns:
        Dictionary containing export statistics including:
        - total_logs: Number of logs exported
        - output_file: Path to the exported file
        - format: Export format used
        - file_size: Size of the exported file in bytes

    Raises:
        LogExtractionAPIError: If export fails
        ValidationError: If parameters are invalid

    Example:
        >>> export_result = api.export_logs_to_file(
        ...     org="my_company",
        ...     start_seq=1,
        ...     log_types=["AUTH", "ADMIN_AUDIT"],
        ...     output_file="/path/to/audit_logs.json",
        ...     format_type="json"
        ... )
        >>> print(f"Exported {export_result['total_logs']} logs")
    """
    import csv
    import json
    from pathlib import Path

    # Validate format
    if format_type not in ["json", "csv", "txt"]:
        raise ValidationError("format_type must be 'json', 'csv', or 'txt'")

    if not output_file:
        raise ValidationError("output_file must be provided")

    # Extract all logs
    all_logs = self.extract_all_logs(
        org=org,
        start_seq=start_seq,
        log_types=log_types,
        end_seq=end_seq,
        batch_size=batch_size,
        start_date=start_date,
        end_date=end_date,
    )

    output_path = Path(output_file)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        if format_type == "json":
            with open(output_path, "w", encoding="utf-8") as f:
                json.dump(all_logs, f, indent=2, default=str)

        elif format_type == "csv":
            if all_logs:
                from silo_sdk.logging.log_schemas import get_fieldnames

                fieldnames = get_fieldnames(log_types, all_logs)
                sanitized_logs = [_sanitize_csv_row(log) for log in all_logs]

                with open(output_path, "w", newline="", encoding="utf-8") as f:
                    writer = csv.DictWriter(
                        f, fieldnames=fieldnames, extrasaction="ignore"
                    )
                    writer.writeheader()
                    writer.writerows(sanitized_logs)

        elif format_type == "txt":
            with open(output_path, "w", encoding="utf-8") as f:
                for i, log in enumerate(all_logs):
                    f.write(f"Log {i + 1}:\n")
                    for key, value in log.items():
                        f.write(f"  {key}: {value}\n")
                    f.write("\n")

        # Get file size
        file_size = output_path.stat().st_size

        self.logger.info(
            "Successfully exported %d logs to %s (%d bytes)",
            len(all_logs),
            output_file,
            file_size,
        )

        return {
            "total_logs": len(all_logs),
            "output_file": str(output_path),
            "format": format_type,
            "file_size": file_size,
        }

    except Exception as e:
        raise LogExtractionAPIError(f"Failed to export logs to file: {e}") from e

group_logs_by_type

group_logs_by_type(logs: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]

Group a flat list of log entries by their type field.

Parameters:

Name	Type	Description	Default
logs	list[dict[str, Any]]	List of log entry dictionaries	required

Returns:

Type	Description
dict[str, list[dict[str, Any]]]	Dictionary mapping log type name to list of entries for that type.
dict[str, list[dict[str, Any]]]	Entries without a type field are grouped under "UNKNOWN".

Example

logs = [ ... {"type": "AUTH", "user": "alice"}, ... {"type": "SESSION", "session_id": "s1"}, ... {"type": "AUTH", "user": "bob"}, ... {"no_type_field": "value"} ... ] grouped = api.group_logs_by_type(logs) len(grouped["AUTH"]) 2 len(grouped["UNKNOWN"]) 1

Source code in silo_sdk/logging/extraction_api.py

@api_tag(MethodType.UTILITY)
def group_logs_by_type(
    self, logs: list[dict[str, Any]]
) -> dict[str, list[dict[str, Any]]]:
    """Group a flat list of log entries by their type field.

    Args:
        logs: List of log entry dictionaries

    Returns:
        Dictionary mapping log type name to list of entries for that type.
        Entries without a type field are grouped under "UNKNOWN".

    Example:
        >>> logs = [
        ...     {"type": "AUTH", "user": "alice"},
        ...     {"type": "SESSION", "session_id": "s1"},
        ...     {"type": "AUTH", "user": "bob"},
        ...     {"no_type_field": "value"}
        ... ]
        >>> grouped = api.group_logs_by_type(logs)
        >>> len(grouped["AUTH"])
        2
        >>> len(grouped["UNKNOWN"])
        1
    """
    grouped: dict[str, list[dict[str, Any]]] = {}

    for entry in logs:
        log_type = entry.get("type", "UNKNOWN")
        if log_type not in grouped:
            grouped[log_type] = []
        grouped[log_type].append(entry)

    return grouped

Log Parsing Utilities

silo_sdk.logging.log_utils provides helpers for working with extracted log data — parsing JSON-encoded fields, building frequency tables, parsing LAUNCHER egress hierarchy strings, and flattening nested structures for CSV export. These functions are importable directly from silo_sdk.logging.

Utilities for parsing and transforming Silo platform log data.

Provides helpers for:

• Parsing JSON-encoded string fields (headers, response_headers)
• Extracting frequency tables (User-Agent, Content-Type)
• Parsing LAUNCHER egress_region hierarchy strings
• Normalizing wire-format log type names to SDK names
• Flattening nested dicts and JSON fields for CSV export

parse_json_field

parse_json_field(log_entry: dict[str, Any], field: str) -> dict[str, Any] | None

Parse a JSON-encoded string field into a dict.

Parameters:

Name	Type	Description	Default
log_entry	dict[str, Any]	Log entry dictionary.	required
field	str	Field name to parse.	required

Returns:

Type	Description
dict[str, Any] | None	Parsed dict, or None if the field is missing, not a string,
dict[str, Any] | None	or not valid JSON.

parse_all_json_fields

parse_all_json_fields(log_entry: dict[str, Any]) -> dict[str, Any]

Parse all known JSON-encoded fields in a log entry.

Replaces headers and response_headers string values with their parsed dict equivalents. Fields that fail to parse are left unchanged.

Parameters:

Name	Type	Description	Default
log_entry	dict[str, Any]	Log entry dictionary.	required

Returns:

Type	Description
dict[str, Any]	New dictionary with parsed JSON fields. The original is not
dict[str, Any]	modified.

extract_user_agents

extract_user_agents(logs: list[dict[str, Any]]) -> Counter[str]

Extract User-Agent frequency table from log entries.

Parses headers (JSON string or dict) and counts the User-Agent value across all entries.

Parameters:

Name	Type	Description	Default
logs	list[dict[str, Any]]	List of log entry dicts (typically URL or ISOLATE_BYPASS).	required

Returns:

Type	Description
Counter[str]	class:~collections.Counter mapping User-Agent string to count.

extract_content_types

extract_content_types(logs: list[dict[str, Any]]) -> Counter[str]

Extract Content-Type frequency from response headers.

Parses response_headers and counts the Content-Type value.

Parameters:

Name	Type	Description	Default
logs	list[dict[str, Any]]	List of log entry dicts (typically URL).	required

Returns:

Type	Description
Counter[str]	class:~collections.Counter mapping Content-Type string to count.

parse_egress_hierarchy

parse_egress_hierarchy(egress_region: str) -> dict[str, str | None]

Parse a LAUNCHER egress_region hierarchy string into components.

The LAUNCHER log type returns egress_region as a slash-separated hierarchy path like::

World /  / North America /  / United States /  / New York, NY

This function splits it into named components.

Parameters:

Name	Type	Description	Default
egress_region	str	Raw hierarchy string from LAUNCHER logs.	required

Returns:

Type	Description
dict[str, str | None]	Dictionary with keys world, region, country, city.
dict[str, str | None]	Missing levels are None.

Example

parse_egress_hierarchy( ... "World / / North America / / United States / / New York, NY" ... ) {'world': 'World', 'region': 'North America', 'country': 'United States', 'city': 'New York, NY'}

normalize_log_type

normalize_log_type(log_entry: dict[str, Any]) -> dict[str, Any]

Convert wire-format type field to SDK name.

The backend returns BLOCKED URL, CASE MANAGER, LOCATION CHANGE, and POST DATA with spaces. This function converts those to the SDK's underscore convention.

Parameters:

Name	Type	Description	Default
log_entry	dict[str, Any]	Log entry dictionary.	required

Returns:

Type	Description
dict[str, Any]	New dictionary with normalized type field. The original
dict[str, Any]	is not modified.

normalize_log_types

normalize_log_types(logs: list[dict[str, Any]]) -> list[dict[str, Any]]

Normalize type fields across a list of log entries.

Parameters:

Name	Type	Description	Default
logs	list[dict[str, Any]]	List of log entry dicts.	required

Returns:

Type	Description
list[dict[str, Any]]	New list with normalized type fields.

expand_json_fields

expand_json_fields(log_entry: dict[str, Any], fields: list[str] | None = None) -> dict[str, Any]

Expand JSON-encoded string fields into dot-notation columns.

For example, a headers field containing '{"User-Agent": "Mozilla/5.0"}' becomes {"headers.User-Agent": "Mozilla/5.0"}.

The original JSON string field is removed.

Parameters:

Name	Type	Description	Default
log_entry	dict[str, Any]	Log entry dictionary.	required
fields	list[str] | None	Fields to expand. Defaults to :data:KNOWN_JSON_FIELDS.	None

Returns:

Type	Description
dict[str, Any]	New dictionary with expanded fields.

flatten_nested_dicts

flatten_nested_dicts(log_entry: dict[str, Any], fields: list[str] | None = None) -> dict[str, Any]

Flatten nested dict fields using dot notation.

For fields that are already dicts (not JSON strings), this flattens them into top-level keys. For example::

{"egress_info": {"protocol": "direct", "connectivity": "datacenter"}}

becomes::

{"egress_info.protocol": "direct", "egress_info.connectivity": "datacenter"}

Parameters:

Name	Type	Description	Default
log_entry	dict[str, Any]	Log entry dictionary.	required
fields	list[str] | None	Specific fields to flatten. If None, flattens all dict-valued fields.	None

Returns:

Type	Description
dict[str, Any]	New dictionary with flattened fields.