Meraki Dashboard

wingpy.cisco.merakidashboard.CiscoMerakiDashboard

CiscoMerakiDashboard(
    *,
    token: str | None = None,
    org_name: str | None = None,
    network_name: str | None = None,
    verify: SSLContext | bool = True,
    timeout: int = 10,
    retries: int = 10
)

Bases: RestApiBaseClass

Interact with the Cisco Meraki Dashboard API.

Parameters:

Name	Type	Description	Default
token	str | None	Bearer token for API authentication. Overrides the environment variable WINGPY_MERAKI_DASHBOARD_TOKEN.	None
org_name	str | None	Default organization name. Overrides the environment variable WINGPY_MERAKI_DASHBOARD_ORG_NAME.	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 CiscoMerakiDashboard
hyperfabric = CiscoMerakiDashboard(
    token=" ",
)
hyperfabric.get("/devices")

Warnings

Not all requests to Meraki Dashboard are suitable for concurrency. Examples include DELETE /networks/{networkId} and POST /organizations/{organizationId}/networks/combine.

For those endpoints, it is recommended to use single-treading for maximum performance.

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

def __init__(
    self,
    *,
    token: str | None = None,
    org_name: str | None = None,
    network_name: str | None = None,
    verify: SSLContext | bool = True,
    timeout: int = 10,
    retries: int = 10,
):
    self.meraki_dashboard_base_url = "https://api.meraki.com/api/v1"
    """The base URL for the Meraki Dashboard API."""

    # Allow parameters to be passed directly or fallback to environment variables

    self.token = token or os.getenv("WINGPY_MERAKI_DASHBOARD_TOKEN")
    """
    The bearer token for authenticating to the Meraki Dashboard API.

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

    self.meraki_dashboard_org_name = org_name or os.getenv(
        "WINGPY_MERAKI_DASHBOARD_ORG_NAME"
    )
    """
    Default organization name.

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

    self.meraki_dashboard_network_name = network_name
    """
    Default network name
    """

    if not self.token:
        raise ValueError(
            "Meraki Dashboard token must be provided either as arguments or environment variables"
        )

    super().__init__(
        base_url=self.meraki_dashboard_base_url,
        auth_lifetime=0,
        auth_refresh_percentage=0,
        verify=verify,
        rate_limit_period=0,
        rate_limit_max_requests=0,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
        backoff_multiplier=1.1,
        timeout=timeout,
        retries=retries,
    )

get

get(
    path: str,
    *,
    params: dict[str, str | list[str]] | None = None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> httpx.Response

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
params	dict[str, str | list[str]] | None	URL query parameters to include in the request. will be added as ?key=value pairs in the URL. Some Meraki Dashboard API endpoints have query parameters where the value data type is "array of strings", ie. productTypes with the /organizations/{organizationId}/networks API endpoint. In those cases, the Meraki Dashboard API server expects the query parameter name to end with []. Example: params={"productTypes[]": ["switch", "camera"]}	None
path_params	dict | None	Replace placeholders like {fabricId} 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
Response	The httpx.Response object from the request.

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

def get(
    self,
    path: str,
    *,
    params: dict[str, str | list[str]] | None = None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> httpx.Response:
    """
    Send an HTTP `GET` request to the specified path.

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

    params : dict[str, str | list[str]] | None, default=None
        URL query parameters to include in the request. will be added as `?key=value` pairs in the URL.

        Some Meraki Dashboard API endpoints have query parameters where the value data type is "array of strings",
        ie. `productTypes` with the `/organizations/{organizationId}/networks` API endpoint.

        In those cases, the Meraki Dashboard API server expects the query parameter name to end with `[]`.

        Example: `params={"productTypes[]": ["switch", "camera"]}`

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

        Will be combined with [`self.path_params`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.path_params] before sending request.

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

        Will be combined with [`self.headers`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.headers] before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [`self.timeout`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.timeout] for a single request.

    Returns
    -------
    httpx.Response
        The [`httpx.Response`](https://www.python-httpx.org/api/#response) object from the request.
    """

    if isinstance(params, dict):
        for param_key, param_value in params.items():
            if isinstance(param_value, list) and param_key[-2:] != "[]":
                raise ValueError(
                    "Invalid query parameter name for 'list of strings'. List based query parameter names must end with []. I.e. {param_key}[] instaed of {param_key}"
                )

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

    return response

get_all

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

Retrieves all pages of data from a JSON endpoint.

Parameters:

Name	Type	Description	Default
path	str	The API endpoint path to send the request to.	required
params	dict[str, str | list[str]] | None	URL query parameters to include in the request. will be added as ?key=value pairs in the URL. Some Meraki Dashboard API endpoints have query parameters where the value data type is "array of strings", ie. productTypes with the /organizations/{organizationId}/networks API endpoint. In those cases, the Meraki Dashboard API server expects the query parameter name to end with []. Example: params={"productTypes[]": ["switch", "camera"]}	None
path_params	dict | None	Replace placeholders like {objectId} 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
page_size	int | None	Set a specific page size. Default value varies on the server side per API endpoint.	None

Returns:

Type	Description
list[dict]	A list of dictionaries, similar to the endpoint-specific key in the JSON responses.

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

def get_all(
    self,
    path: str,
    *,
    params: dict[str, str | list[str]] | None = None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
    page_size: int | None = None,
) -> list:
    """
    Retrieves all pages of data from a JSON endpoint.

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

    params : dict[str, str | list[str]] | None, default=None
        URL query parameters to include in the request. will be added as `?key=value` pairs in the URL.

        Some Meraki Dashboard API endpoints have query parameters where the value data type is "array of strings",
        ie. `productTypes` with the `/organizations/{organizationId}/networks` API endpoint.

        In those cases, the Meraki Dashboard API server expects the query parameter name to end with `[]`.

        Example: `params={"productTypes[]": ["switch", "camera"]}`

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

        Will be combined with [`self.path_params`][wingpy.cisco.catalystcenter.CiscoCatalystCenter.path_params] before sending request.

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

        Will be combined with [`self.headers`][wingpy.cisco.catalystcenter.CiscoCatalystCenter.headers] before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [`self.timeout`][wingpy.cisco.catalystcenter.CiscoCatalystCenter.timeout] for a single request.

    page_size : int | None, default=None
        Set a specific page size. Default value varies on the server side per API endpoint.

    Returns
    -------
    list[dict]
        A list of dictionaries, similar to the endpoint-specific key in the JSON responses.
    """

    merged_params = {}

    if isinstance(params, dict):
        merged_params.update(params)

    if page_size:
        merged_params["perPage"] = page_size

    first_page = self.get(
        path=path,
        params=merged_params,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
    )

    result = first_page.json()

    link_header = first_page.headers.get("link", "")

    while "rel=next" in link_header:
        matches = re.findall(r"<(\S+?)>; rel=(\S+?)(?:,|$)", link_header)
        next_url = None
        for url, rel in matches:
            if rel == "next":
                next_url = url

        if next_url is None or self.base_url not in next_url:
            raise RuntimeError(
                f"Pagination error: valid 'next' page URL not found in link header. {link_header}"
            )

        if self.base_url != next_url[: len(self.base_url)]:  # pragma: nocover
            raise RuntimeError(
                f"Pagination error: Invalid 'next' page URL. {next_url}"
            )

        next_path = next_url[len(self.base_url) :]

        page = self.get(
            path=next_path,
            path_params=path_params,
            headers=headers,
            timeout=timeout,
        )

        result += page.json()

        link_header = page.headers.get("link", "")

    return result

post

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

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 {fabricId} 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
Response	The httpx.Response object from the request.

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

def post(
    self,
    path: str,
    *,
    data: str | dict | list | None,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> httpx.Response:
    """
    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 `{fabricId}` in the URL path with actual values.

        Will be combined with [`self.path_params`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.path_params] before sending request.

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

        Will be combined with [`self.headers`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.headers] before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [`self.timeout`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.timeout] for a single request.

    Returns
    -------
    httpx.Response
        The [`httpx.Response`](https://www.python-httpx.org/api/#response) 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,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> httpx.Response

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	Request payload as JSON string or Python list/dict object.	required
path_params	dict | None	Replace placeholders like {fabricId} 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
Response	The httpx.Response object from the request.

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

def put(
    self,
    path: str,
    *,
    data: str | dict | list,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> httpx.Response:
    """
    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
        Request payload as JSON string or Python list/dict object.

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

        Will be combined with [`self.path_params`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.path_params] before sending request.

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

        Will be combined with [`self.headers`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.headers] before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [`self.timeout`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.timeout] for a single request.

    Returns
    -------
    httpx.Response
        The [`httpx.Response`](https://www.python-httpx.org/api/#response) 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

delete

delete(
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None
) -> httpx.Response

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 {fabricId} 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
Response	The httpx.Response object from the request.

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

def delete(
    self,
    path: str,
    *,
    path_params: dict | None = None,
    headers: dict | None = None,
    timeout: int | None = None,
) -> httpx.Response:
    """
    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 `{fabricId}` in the URL path with actual values.

        Will be combined with [`self.path_params`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.path_params] before sending request.

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

        Will be combined with [`self.headers`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.headers] before sending request.

    timeout : int | None, default=None
        Override the standard timeout timer [`self.timeout`][wingpy.cisco.merakidashboard.CiscoMerakiDashboard.timeout] for a single request.

    Returns
    -------
    httpx.Response
        The [`httpx.Response`](https://www.python-httpx.org/api/#response) 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

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 instance-attribute

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 property

is_authenticated

Check if the client is authenticated.

timeout instance-attribute

timeout: int = timeout

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

MAX_CONNECTIONS class-attribute instance-attribute

MAX_CONNECTIONS = 4

The maximum number of concurrent connections opened to Meraki Dashboard.

1 connection will be used for general synchronous requests.

3 connections will be used for parallel asynchronous requests.

RETRY_RESPONSES class-attribute instance-attribute

RETRY_RESPONSES = [
    HttpResponsePattern(
        status_codes=[400],
        methods=["DELETE"],
        content_patterns=[
            compile(
                '{"errors":\\["Unable to destroy network N_\\d+. This may be due to concurrent requests to delete or combine networks\\."\\]}'
            )
        ],
    )
]

The standard HTTP 429 status code indicates that the user has sent too many requests in a given amount of time, and is being rate limited.

For certain endpoints that require asynchronous background processing by Cisco Meraki Dashboard a HTTP 400 will be returned instead.

These responses will be retried.

headers instance-attribute

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 instance-attribute

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.