API Reference

Sync

Sync(
    url: str,
    *,
    filter: Optional[str] = None,
    projection: Optional[Iterable[str]] = None,
    start: Union[datetime, int, float, None] = None,
    user_agent: Optional[str] = None,
    allow_insecure: bool = False,
    **kwargs: Any,
)

Synchronize events from sharing API.

Reads every event matching the given conditions in database insertion order, so that a caller can keep its own copy of the events up to date without missing any.

The name refers to that synchronization, not to the sharing API's synchronous endpoints. This class queries the asynchronous endpoints, as Query does.

Parameters:

Name	Type	Description	Default
`url`	`str`	Sharing API url, must be an `https` url and must include `apikey` query parameter.	required
`filter`	`Optional[str]`	Rulelang filter.	`None`
`projection`	`Optional[Iterable[str]]`	List of event field names to include in the results. Note that the list of keys provided by the server can only be limited by this parameter.	`None`
`start`	`Union[datetime, int, float, None]`	Start time for the initial query. Can also be set with seek(). Default value `None` means current time. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.	`None`
`user_agent`	`Optional[str]`	Value appended to the `User-Agent` request header.	`None`
`allow_insecure`	`bool`	Accept a plain `http` url. The api key is sent on every request, so it is then transmitted in cleartext. Intended for development servers only.	`False`

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def __init__(
    self,
    url: str,
    *,
    filter: Optional[str] = None,
    projection: Optional[Iterable[str]] = None,
    start: Union[datetime, int, float, None] = None,
    user_agent: Optional[str] = None,
    allow_insecure: bool = False,
    **kwargs: Any,
):
    """
    Initialize Sync class.

    Args:
        url: Sharing API url, must be an `https` url and must include
            `apikey` query parameter.
        filter: Rulelang filter.
        projection: List of event field names to include in the results. Note that the list of keys provided by the server can only be limited by this parameter.
        start: Start time for the initial query. Can also be set with seek(). Default value `None` means current time. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.
        user_agent: Value appended to the `User-Agent` request header.
        allow_insecure: Accept a plain `http` url. The api key is sent on every request, so it is then transmitted in cleartext. Intended for development servers only.
    """

    # Check args
    if not isinstance(url, str):
        raise TypeError(f"url must be string not {type(url)}")

    if not (filter is None or isinstance(filter, str)):
        raise TypeError(f"filter must be string or None, not {type(filter)}")

    projection = _validate_projection(projection)

    client_kwargs = _validate_client_kwargs(user_agent, allow_insecure)

    if kwargs:
        raise ValueError(f"Unknown parameter(s) {tuple(kwargs.keys())}")

    # Initialize client
    self.api_client = _ApiClient(url, **client_kwargs)

    invalid_qps_in_url = (
        self.api_client.urls.qp.keys() - self.allowed_user_provided_qps
    )
    if invalid_qps_in_url:
        raise ConfigError(f"Invalid qps in url: {invalid_qps_in_url}")

    self.qp = {
        "filter": filter,
        "projection": projection,
        "sort": "_id",
    }

    self.start = self._seek(start)

read

read(
    *,
    token: Optional[str] = None,
    pagesize: int = 1000,
    timeout: Optional[float] = 600,
) -> SyncReadResponse

Read the next batch of events to synchronize.

Events are returned sorted by insertion time.

Parameters:

Name	Type	Description	Default
`token`	`Optional[str]`	Continuation token. If not given, the start time set earlier is used.	`None`
`pagesize`	`int`	Maximum number of events to return.	`1000`
`timeout`	`Optional[float]`	Timeout for the operation in [s].	`600`

Returns:

Type	Description
`SyncReadResponse`	Batch of events and a continuation token for the next query.

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def read(
    self,
    *,
    token: Optional[str] = None,
    pagesize: int = 1000,
    timeout: Optional[float] = 600,
) -> SyncReadResponse:
    """Read the next batch of events to synchronize.

    Events are returned sorted by insertion time.

    Args:
        token: Continuation token. If not given, the start time set earlier is used.
        pagesize: Maximum number of events to return.
        timeout: Timeout for the operation in [s].

    Returns:
        Batch of events and a continuation token for the next query.
    """
    if not (token is None or isinstance(token, str)):
        raise TypeError(f"token must be string or None, not {type(token)}")

    if not isinstance(pagesize, int):
        raise TypeError(f"pagesize must be int not {type(pagesize)}")

    if not (timeout is None or isinstance(timeout, (int, float))):
        raise TypeError(f"timeout must be float or None, not {type(timeout)}")

    qp = _remove_none_values(
        {
            **self.qp,
            **{
                "token": token,
                "limit": pagesize if pagesize != 0 else None,
            },
        }
    )

    if "token" not in qp:
        qp["start"] = self.start

    resp = self.api_client.async_query(qp, timeout=timeout)

    try:
        token = resp.headers["x-next-token"]
        has_more = True
    except KeyError:
        token = resp.headers.get("x-last-inserted-token", None)
        has_more = False

    return SyncReadResponse(
        _parse_events(resp),
        token,
        has_more,
    )

seek

seek(ts: Union[datetime, int, float, None]) -> None

Set sync start to specific time.

Parameters:

Name	Type	Description	Default
`ts`	`Union[datetime, int, float, None]`	See start in init.	required

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def seek(self, ts: Union[datetime, int, float, None]) -> None:
    """Set sync start to specific time.

    Args:
        ts: See start in __init__.
    """
    self.start = self._seek(ts)

SyncReadResponse `dataclass`

SyncReadResponse(events: list[Event], token: Optional[str], has_more: bool)

Result of a single Sync.read() call.

Attributes:

Name	Type	Description
`events`	`list[Event]`	The batch of events, in database insertion order.
`token`	`Optional[str]`	Continuation token to pass to the next `read()` call. This is the next token when more events are available, and the last inserted token otherwise. It is `None` only when the server reported no position at all, in which case the previously held token should be kept.
`has_more`	`bool`	Whether more events already exist in the database at the moment. When `False`, the caller has reached the end of the stream and should wait before reading again.

Query

Query(
    url: str,
    *,
    user_agent: Optional[str] = None,
    allow_insecure: bool = False,
    **kwargs: Any,
)

Query events from sharing API.

Parameters:

Name	Type	Description	Default
`url`	`str`	Sharing API url, must be an `https` url and must include `apikey` query parameter.	required
`user_agent`	`Optional[str]`	Value appended to the `User-Agent` request header.	`None`
`allow_insecure`	`bool`	Accept a plain `http` url. The api key is sent on every request, so it is then transmitted in cleartext. Intended for development servers only.	`False`

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def __init__(
    self,
    url: str,
    *,
    user_agent: Optional[str] = None,
    allow_insecure: bool = False,
    **kwargs: Any,
):
    """
    Initialize Query class.

    Args:
        url: Sharing API url, must be an `https` url and must include
            `apikey` query parameter.
        user_agent: Value appended to the `User-Agent` request header.
        allow_insecure: Accept a plain `http` url. The api key is sent on every request, so it is then transmitted in cleartext. Intended for development servers only.
    """

    # Check args
    if not isinstance(url, str):
        raise TypeError(f"url must be string not {type(url)}")

    client_kwargs = _validate_client_kwargs(user_agent, allow_insecure)

    if kwargs:
        raise ValueError(f"Unknown parameter(s) {tuple(kwargs.keys())}")

    # Initialize client
    self.api_client = _ApiClient(url, **client_kwargs)

    invalid_qps_in_url = (
        self.api_client.urls.qp.keys() - self.allowed_user_provided_qps
    )
    if invalid_qps_in_url:
        raise ConfigError(f"Invalid qps in url: {invalid_qps_in_url}")

query

query(
    *,
    filter: Optional[str] = None,
    projection: Optional[Iterable[str]] = None,
    start: Union[datetime, int, float, None] = None,
    end: Union[datetime, int, float, None] = None,
    reverse: bool = False,
    max_events: int = 0,
    timeout: Optional[float] = 600,
    **kwargs: Any,
) -> Iterator[Event]

Query sharing API.

Events are returned sorted by timestamp.

Parameters:

Name	Type	Description	Default
`filter`	`Optional[str]`	Rulelang filter.	`None`
`projection`	`Optional[Iterable[str]]`	List of event field names to include in the results. Note that the list of keys provided by the server can only be limited by this parameter.	`None`
`start`	`Union[datetime, int, float, None]`	Start time for the query. Default value `None` means forever. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.	`None`
`end`	`Union[datetime, int, float, None]`	End time for the query. Default value `None` means current time. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.	`None`
`reverse`	`bool`	Return events in reverse order.	`False`
`max_events`	`int`	Maximum number of events to generate.	`0`
`timeout`	`Optional[float]`	Timeout for one sharing API query fetching more events in [s].	`600`

Returns:

Type	Description
`Iterator[Event]`	Generator yielding events matching the query.

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def query(
    self,
    *,
    filter: Optional[str] = None,
    projection: Optional[Iterable[str]] = None,
    start: Union[datetime, int, float, None] = None,
    end: Union[datetime, int, float, None] = None,
    reverse: bool = False,
    max_events: int = 0,
    timeout: Optional[float] = 600,
    **kwargs: Any,
) -> Iterator[Event]:
    """Query sharing API.

    Events are returned sorted by timestamp.

    Args:
        filter: Rulelang filter.
        projection: List of event field names to include in the results. Note that the list of keys provided by the server can only be limited by this parameter.
        start: Start time for the query. Default value `None` means forever. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.
        end: End time for the query. Default value `None` means current time. Positive numbers are interpret as epoch time. Non-positive numbers are interpret as that many seconds in the past.
        reverse: Return events in reverse order.
        max_events: Maximum number of events to generate.
        timeout: Timeout for one sharing API query fetching more events in [s].

    Returns:
        Generator yielding events matching the query.
    """
    if not (filter is None or isinstance(filter, str)):
        raise TypeError(f"filter must be string or None, not {type(filter)}")

    projection = _validate_projection(projection)

    if not (start is None or isinstance(start, (int, float, datetime))):
        raise TypeError(
            f"start must be int, float, datetime or None, not {type(start)}"
        )

    if not (end is None or isinstance(end, (int, float, datetime))):
        raise TypeError(
            f"end must be int, float, datetime or None, not {type(end)}"
        )

    if not isinstance(reverse, bool):
        raise TypeError(f"reverse must be bool, not {type(bool)}")

    if not isinstance(max_events, int):
        raise TypeError(f"max_events must be int, not {type(max_events)}")

    if not (timeout is None or isinstance(timeout, (int, float))):
        raise TypeError(f"timeout must be float or None, not {type(timeout)}")

    pagesize = kwargs.pop("pagesize", 1000)
    if not isinstance(pagesize, int):
        raise TypeError(f"pagesize must be int, not {type(kwargs.get('pagesize'))}")

    if kwargs:
        raise ValueError(f"Unknown parameter(s) {tuple(kwargs.keys())}")

    qp = _remove_none_values(
        {
            "filter": filter,
            "projection": projection,
            "start": _build_start_end(start),
            "end": _build_start_end(end),
            "reverse": "" if reverse else None,
            "limit": pagesize,
        }
    )

    return self._iter_events(qp, max_events, timeout)

query

query(url: str, **kwargs: Any) -> Iterable[Event]

Shortcut to Query(url).query().

Source code in src/arcticsecurity/sharing_api/sharing_api.py

def query(url: str, **kwargs: Any) -> Iterable[Event]:
    """Shortcut to Query(url).query()."""
    return Query(url).query(**kwargs)

Types

Event `module-attribute`

Event = dict[str, Union[str, list[str]]]

A single event, mapping a field name to its value or list of values.

Exception classes

Error

Error(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Exception

Base class for all errors.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

ConfigError

ConfigError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Error

User error in query configuration.

This error may be raised by this library or by the server.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

AuthError

AuthError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: ConfigError

The server rejected the api key.

The share url is missing, has an expired or revoked api key, or the key does not grant access to the requested share. Retrying will not help until the url is fixed.

This is a subclass of ConfigError, since it always indicates a problem with the configured share url.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

InvalidTokenError

InvalidTokenError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Error

The server responded with "invalid token" error.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

ServerError

ServerError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Error

The server responded with a error message.

This results from a server which doesn't operate properly at the moment.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

NetworkError

NetworkError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Error

Error communicating with the sharing API server.

This results from error communicating the with the sharing API server.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

TimeoutError

TimeoutError(*args: Any, url: Optional[str] = None, **kwargs: Any)

Bases: Error

Query to sharing API server timed out.

The three-phase query to the sharing API exceeded the configured timeout.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(self, *args: Any, url: Optional[str] = None, **kwargs: Any):
    self.url = url
    super().__init__(*args, **kwargs)

Retry

Retry(*args: Any, after: Union[str, int, float, None] = None, **kwargs: Any)

Bases: NetworkError

Network error that should be retried.

This is raised from errors that are likely transient and may succeed if retried after some time. Attributes: after: Number of seconds the server asked the client to wait before retrying, or None if the server did not say (or said something unparseable). This value is advisory and is not bounded; the caller decides how long it is willing to wait.

Source code in src/arcticsecurity/sharing_api/errors.py

def __init__(
    self, *args: Any, after: Union[str, int, float, None] = None, **kwargs: Any
):
    super().__init__(*args, **kwargs)

    self.after = parse_retry_after(after)

API Reference

Sync

Sync

read

seek

SyncReadResponse dataclass

Query

Query

query

query

Types

Event module-attribute

Exception classes

Error

ConfigError

AuthError

InvalidTokenError

ServerError

NetworkError

TimeoutError

Retry

SyncReadResponse `dataclass`

Event `module-attribute`