Secure Network Analytics

wingpy.cisco.sna.CiscoSNA

CiscoSNA(
    *,
    base_url: str | None = None,
    username: str | None = None,
    password: str | None = None,
    tenant_name: str | None = None,
    verify: SSLContext | bool = True,
    timeout: int = 10,
    retries: int = 3
)

Bases: RestApiBaseClass

Interact with the Cisco Secure Network Analytics API.

Parameters:

Name	Type	Description	Default
base_url	str | None	Base URL of the API including https://. Overrides the environment variable WINGPY_SNA_BASE_URL.	None
username	str | None	Username for API authentication. Overrides the environment variable WINGPY_SNA_USERNAME.	None
password	str | None	Password for API authentication. Overrides the environment variable WINGPY_SNA_PASSWORD.	None
tenant_name	str | None	The name of the tenant to query. Overrides the environment variable WINGPY_SNA_TENANT_NAME. If neither the parameter or environment variable are set, the first or only available tenant is used.	None
verify	bool | SSLContext	Boolean values will enable or disable the default SSL verification. Use an ssl.SSLContext to specify custom Certificate Authority.	True
timeout	int	Number of seconds to wait for HTTP responses before raising httpx.TimeoutException exception.	10
retries	int	Number of failed HTTP attempts allowed before raising httpx.HTTPStatusError exception.	3

Examples:

from wingpy import CiscoSNA
sna = CiscoSNA(
    base_url="https://sna.example.com",
    username="example_username",
    password="example_password",
)
tags = sna.get_all("/sw-reporting/v1/tenants/{tenantId}/customHosts/tags")
print(f"Retrieved {len(tags)} tags")

Source code in src/wingpy/cisco/sna.py

def __init__(
    self,
    *,
    base_url: str | None = None,
    username: str | None = None,
    password: str | None = None,
    tenant_name: str | None = None,
    verify: SSLContext | bool = True,
    timeout: int = 10,
    retries: int = 3,
):
    # Allow parameters to be passed directly or fallback to environment variables
    self.sna_url = base_url or os.getenv("WINGPY_SNA_BASE_URL")
    """
    The base URL for the Cisco Secure Network Analytics API.

    If not provided, it will be read from the environment variable `WINGPY_SNA_BASE_URL`.
    """

    self.username = username or os.getenv("WINGPY_SNA_USERNAME")
    """
    The username for authentication.

    If not provided, it will be read from the environment variable `WINGPY_SNA_USERNAME`.
    """

    self.password = password or os.getenv("WINGPY_SNA_PASSWORD")
    """
    The password for authentication.

    If not provided, it will be read from the environment variable `WINGPY_SNA_PASSWORD`.
    """

    self.tenant_name = tenant_name or os.getenv("WINGPY_SNA_TENANT_NAME")
    """
    The name of the tenant to query.

    If not provided, it will be read from the environment variable `WINGPY_SNA_TENANT_NAME`.

    If neither the parameter or environment variable are set, the first or only available tenant is used.
    """

    if not self.sna_url:
        raise ValueError(
            "Cisco Secure Network Analytics base_url must be provided either as argument or environment variable"
        )

    if not self.username or not self.password:
        raise ValueError(
            "Cisco SNA username and password must be provided either as arguments or environment variables"
        )

    self.token: str | None = None
    """
    The authentication token for Cisco SNA.
    """

    self.version: Version = Version("0.0")
    """
    The version of the Cisco Secure Network Analytics API.
    """

    super().__init__(
        base_url=self.sna_url,
        auth_lifetime=1800,
        auth_refresh_percentage=0.9,
        http2=True,
        verify=verify,
        timeout=timeout,
        retries=retries,
        headers={
            "Content-Type": "application/json",
            # `Accept` header is not supported
        },
    )

    self.path_params: dict = {}
    """
    A dictionary of path parameters to be used in the API path of each request.

    These parameters will be merged with any `path_params` dict passed to the request.

    `{tenantId}` is automatically resolved based on the [`tenant_name`](https://wingpy.automation.wingmen.dk/api/sna/#wingpy.cisco.sna.CiscoSNA) passed to the constructor.
    """

get

get(
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> ResponseMapping | ResponseSequence

Send an HTTP GET request to the specified path.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
path_params	dict | None	Replace placeholders like {tagId} in the URL path with actual values. Will be combined with self.path_params before sending request.	None
headers	dict | None	HTTP headers to be sent with the request. Will be combined with self.headers before sending request.	None
timeout	int | None	Override the standard timeout timer self.timeout for a single request.	None

Returns:

Type	Description
ResponseMapping | ResponseSequence	The ResponseMapping or ResponseSequence object from the request.

Source code in src/wingpy/cisco/sna.py

def get(
    self,
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> ResponseMapping | ResponseSequence:
    """
    Send an HTTP `GET` request to the specified path.

    Parameters
    ----------
    path : str
        The API endpoint path to send the request to.

    path_params : dict | None, default=None
        Replace placeholders like `{tagId}` in the URL path with actual values.

        Will be combined with [self.path_params](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.path_params) before sending request.

    headers : dict | None, default=None
        HTTP headers to be sent with the request.

        Will be combined with [self.headers](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.headers) before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [self.timeout](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.timeout) for a single request.

    Returns
    -------
    ResponseMapping | ResponseSequence
        The [`ResponseMapping`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseMapping) or [`ResponseSequence`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseSequence)
        object from the request.
    """

    response = self.request(
        "GET",
        path,
        data=None,
        params=None,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
        is_auth_endpoint=False,
        auth=None,
    )
    return response

get_all

get_all(
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> list

Secure Network Analytics does not use pagination.

Response data is placed in the data element by the API.

Some list-data is nested further by the API In those cases the actual list is extracted and returned.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
path_params	dict | None	Replace placeholders like {tagId} in the URL path with actual values. Will be combined with self.path_params before sending request.	None
headers	dict | None	HTTP headers to be sent with the request. Will be combined with self.headers before sending request.	None
timeout	int | None	Override the standard timeout timer self.timeout for a single request.	None

Returns:

Type	Description
list[dict]	A list-version of returned data.

Source code in src/wingpy/cisco/sna.py

def get_all(
    self,
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> list:
    """
    Secure Network Analytics does not use pagination.

    Response data is placed in the `data` element by the API.

    Some list-data is nested further by the API In those
    cases the actual list is extracted and returned.

    Parameters
    ----------
    path : str
        The API endpoint path to send the request to.

    path_params : dict | None, default=None
        Replace placeholders like `{tagId}` in the URL path with actual values.

        Will be combined with [self.path_params](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.path_params) before sending request.

    headers : dict | None, default=None
        HTTP headers to be sent with the request.

        Will be combined with [self.headers](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.headers) before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [self.timeout](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.timeout) for a single request.

    Returns
    -------
    list[dict]
        A list-version of returned data.
    """

    response = self.get(
        path,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
    )

    if isinstance(response, ResponseSequence):
        result = list(response)
    elif isinstance(response, ResponseMapping):
        if isinstance(response["data"], list):
            result = response["data"]
        elif isinstance(response["data"], dict):
            # Some endpoints include a summary as the first element
            # Use the last element
            *_, result = iter(response["data"].values())

    else:
        error = UnexpectedPayloadError(
            "Unable to retrieve list-based data", response=response
        )
        log_exception(error)
        raise error

    logger.debug(f"Received {len(result)} total items from {path}")
    return result

post

post(
    path: str,
    *,
    data: str | dict | list | None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> ResponseMapping | ResponseSequence

Send an HTTP POST request to the specified path.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
data	str | dict | list | None	Request payload as JSON string or Python list/dict object.	required
path_params	dict | None	Replace placeholders like {tagId} in the URL path with actual values. Will be combined with self.path_params before sending request.	None
headers	dict | None	HTTP headers to be sent with the request. Will be combined with self.headers before sending request.	None
timeout	int | None	Override the standard timeout timer self.timeout for a single request.	None

Returns:

Type	Description
ResponseMapping | ResponseSequence	The ResponseMapping or ResponseSequence object from the request.

Source code in src/wingpy/cisco/sna.py

def post(
    self,
    path: str,
    *,
    data: str | dict | list | None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> ResponseMapping | ResponseSequence:
    """
    Send an HTTP `POST` request to the specified path.

    Parameters
    ----------
    path : str
        The API endpoint path to send the request to.

    data : str | dict | list | None
        Request payload as JSON string or Python list/dict object.

    path_params : dict | None, default=None
        Replace placeholders like `{tagId}` in the URL path with actual values.

        Will be combined with [self.path_params](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.path_params) before sending request.

    headers : dict | None, default=None
        HTTP headers to be sent with the request.

        Will be combined with [self.headers](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.headers) before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [self.timeout](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.timeout) for a single request.

    Returns
    -------
    ResponseMapping | ResponseSequence
        The [`ResponseMapping`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseMapping) or [`ResponseSequence`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseSequence)
        object from the request.
    """

    response = self.request(
        "POST",
        path,
        data=data,
        params=None,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
        is_auth_endpoint=False,
        auth=None,
    )
    return response

put

put(
    path: str,
    *,
    data: str | dict | list | None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> ResponseMapping | ResponseSequence

Send an HTTP PUT request to the specified path.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
data	str | dict | list | None	Request payload as JSON string or Python list/dict object.	required
path_params	dict | None	Replace placeholders like {tagId} in the URL path with actual values. Will be combined with self.path_params before sending request.	None
headers	dict | None	HTTP headers to be sent with the request. Will be combined with self.headers before sending request.	None
timeout	int | None	Override the standard timeout timer self.timeout for a single request.	None

Returns:

Type	Description
ResponseMapping | ResponseSequence	The ResponseMapping or ResponseSequence object from the request.

Source code in src/wingpy/cisco/sna.py

def put(
    self,
    path: str,
    *,
    data: str | dict | list | None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> ResponseMapping | ResponseSequence:
    """
    Send an HTTP `PUT` request to the specified path.

    Parameters
    ----------
    path : str
        The API endpoint path to send the request to.

    data : str | dict | list | None
        Request payload as JSON string or Python list/dict object.

    path_params : dict | None, default=None
        Replace placeholders like `{tagId}` in the URL path with actual values.

        Will be combined with [self.path_params](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.path_params) before sending request.

    headers : dict | None, default=None
        HTTP headers to be sent with the request.

        Will be combined with [self.headers](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.headers) before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [self.timeout](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.timeout) for a single request.

    Returns
    -------
    ResponseMapping | ResponseSequence
        The [`ResponseMapping`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseMapping) or [`ResponseSequence`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseSequence)
        object from the request.
    """

    response = self.request(
        "PUT",
        path,
        data=data,
        params=None,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
        is_auth_endpoint=False,
        auth=None,
    )
    return response

patch

patch(*args, **kwargs) -> None

HTTP PATCH is not supported by Cisco Secure Network Analytics

Raises:

Type	Description
UnsupportedMethodError

Source code in src/wingpy/cisco/sna.py

def patch(self, *args, **kwargs) -> None:  # ignore: type
    """
    !!! failure "HTTP PATCH is not supported by Cisco Secure Network Analytics"

    Raises
    ------
    UnsupportedMethodError
    """
    error = UnsupportedMethodError(client=self, method="PATCH")
    log_exception(error)
    raise error

delete

delete(
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> ResponseMapping | ResponseSequence

Send an HTTP DELETE request to the specified path.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
path_params	dict | None	Replace placeholders like {tagId} in the URL path with actual values. Will be combined with self.path_params before sending request.	None
headers	dict | None	HTTP headers to be sent with the request. Will be combined with self.headers before sending request.	None
timeout	int | None	Override the standard timeout timer self.timeout for a single request.	None

Returns:

Type	Description
ResponseMapping | ResponseSequence	The ResponseMapping or ResponseSequence object from the request.

Source code in src/wingpy/cisco/sna.py

def delete(
    self,
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> ResponseMapping | ResponseSequence:
    """
    Send an HTTP `DELETE` request to the specified path.

    Parameters
    ----------
    path : str
        The API endpoint path to send the request to.

    path_params : dict | None, default=None
        Replace placeholders like `{tagId}` in the URL path with actual values.

        Will be combined with [self.path_params](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.path_params) before sending request.

    headers : dict | None, default=None
        HTTP headers to be sent with the request.

        Will be combined with [self.headers](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.headers) before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [self.timeout](https://wingpy.automation.wingmen.dk/api/secure-network-analytics/#wingpy.cisco.sna.CiscoSNA.timeout) for a single request.

    Returns
    -------
    ResponseMapping | ResponseSequence
        The [`ResponseMapping`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseMapping) or [`ResponseSequence`](https://wingpy.automation.wingmen.dk/api/response/#wingpy.response.ResponseSequence)
        object from the request.
    """

    response = self.request(
        "DELETE",
        path,
        data=None,
        params=None,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
        is_auth_endpoint=False,
        auth=None,
    )
    return response

wait_for

wait_for(
    func: Callable,
    path: str,
    on_complete: Callable | None = None,
    **kwargs
) -> str | Any

Provides easy access to the result of job generating APIs. These start asycronous jobs in the background where status can be queried periodically. When the job is done a new URL path is returned or passed to another callable.

Examples:

By default a URL path for the final result is returned. You can then use get or get_all with it.

query = {
    "startDateTime": "2026-08-11T08:00:00Z",
    "endDateTime": "2026-08-11T08:00:00Z",
}
flow_result_path = sna.wait_for(
    sna.post,
    "/sw-reporting/v2/tenants/{tenantId}/flows/queries",
    data=query,
)
flows = sna.get_all(flow_result_path)

Using the on_complete hook automatically runs that function with the returned path:

query = {
    "startTime": "2026-08-11T08:00:00.000",
    "endTime": "2026-08-11T09:00:00.000",
}
top_hosts = sna.wait_for(
    sna.post,
    "/sw-reporting/v1/tenants/{tenantId}/flow-reports/top-hosts/queries",
    data=query,
    on_complete=sna.get_all,
)

Parameters:

Name	Type	Description	Default
func	Callable	The initial method that creates the job.	required
path	str	The API endpoint path to send the request to.	required
on_complete	Callable	A method to call with the job result URL path after waiting for it to become ready.	None

Other Parameters:

Name	Type	Description
**kwargs		Any keyword parameters passed on to the func, ie. headers, params, ...

Returns:

Type	Description
str	URL path for accessing the final result.
Any	Result from on_complete Callable passed through.

Source code in src/wingpy/cisco/sna.py

def wait_for(
    self,
    func: Callable,
    path: str,
    on_complete: Callable | None = None,
    **kwargs,
) -> str | Any:
    """
    Provides easy access to the result of job generating APIs.
    These start asycronous jobs in the background where status can be queried periodically.
    When the job is done a new URL path is returned or passed to another callable.

    Examples
    --------
    By default a URL path for the final result is returned. You can then use get or get_all with it.
    ```python
    query = {
        "startDateTime": "2026-08-11T08:00:00Z",
        "endDateTime": "2026-08-11T08:00:00Z",
    }
    flow_result_path = sna.wait_for(
        sna.post,
        "/sw-reporting/v2/tenants/{tenantId}/flows/queries",
        data=query,
    )
    flows = sna.get_all(flow_result_path)
    ```

    Using the on_complete hook automatically runs that function with the returned path:
    ```python
    query = {
        "startTime": "2026-08-11T08:00:00.000",
        "endTime": "2026-08-11T09:00:00.000",
    }
    top_hosts = sna.wait_for(
        sna.post,
        "/sw-reporting/v1/tenants/{tenantId}/flow-reports/top-hosts/queries",
        data=query,
        on_complete=sna.get_all,
    )
    ```

    Parameters
    ----------
    func : Callable
        The initial method that creates the job.

    path : str
        The API endpoint path to send the request to.

    on_complete : Callable
        A method to call with the job result URL path after waiting for it to become ready.

    Other parameters
    ----------------

    **kwargs :
        Any keyword parameters passed on to the func, ie. headers, params, ...

    Returns
    -------
    str
        URL path for accessing the final result.

    Any
        Result from on_complete Callable passed through.
    """

    response = func(path, **kwargs)

    if "/v1/" in path and "/queries" in path and response.status_code == 200:
        result_path = self._get_query_result_path_v1(response)
    elif (
        "/v2/" in path
        and "/queries" in path
        and response.status_code == 201
        and "location" in response.headers
    ):
        result_path = self._get_query_result_path_v2(response)
    else:
        error = InvalidJobError(
            "Unable to determine API job type.",
            response=response,
        )
        log_exception(error)
        raise error
    if on_complete:
        return on_complete(result_path)
    else:
        return result_path

authenticate

authenticate() -> None

Executes the API-specific authentication process and records timestamps for session tracking.

Notes

Authentication will automatically be carried out just-in-time.

Only call this method directly if you need to authenticate proactively, outside of normal request flow.

Source code in src/wingpy/base.py

def authenticate(self) -> None:
    """
    Executes the API-specific authentication process and records timestamps
    for session tracking.

    Notes
    ----
    Authentication will automatically be carried out just-in-time.

    Only call this method directly if you need to authenticate proactively,
    outside of normal request flow.
    """

    # Authenticate
    logger.debug("Authenticating and recording token lifetime")
    auth_response = self._authenticate()

    # Record the time of authentication
    self.auth_timestamp = arrow.utcnow()

    self._after_auth(auth_response=auth_response)

tasks

tasks: TaskRunner = TaskRunner(max_workers=max_workers)

Manages concurrent requests to the API server.

The number of concurrent requests is limited by the MAX_CONNECTIONS property:

• 1 connection is reserved for the main thread used for authentication and synchronous requests.
• The remaining connections are used for concurrent requests.

See Also

wingpy.scheduling.TaskRunner Schedule and run asynchronous tasks in parallel.

is_authenticated

is_authenticated: bool

Check if the client is authenticated.

timeout

timeout: int = timeout

The timeout in seconds for each request to the API server.

MAX_CONNECTIONS

MAX_CONNECTIONS = 10

The maximum number of concurrent connections opened to the Cisco Secure Network Analytics API.

1 connection will be used for general synchronous requests.

9 connections will be used for parallel asynchronous requests.

RETRY_RESPONSES

RETRY_RESPONSES = []

No explicit retry reponses are defined for Cisco Secure Network Analytics.

headers

headers: dict = headers or {}

A dictionary of HTTP headers to be sent with each request. These headers will be merged with any headers dict passed to an individual request.

path_params

path_params: dict = {}

A dictionary of path parameters to be used in the API path of each request.

These parameters will be merged with any path_params dict passed to the request.

{tenantId} is automatically resolved based on the tenant_name passed to the constructor.