API Reference¶

async def read_raw(
    self,
    object_type: str,
    *,
    params: Mapping[str, str] | None = None,
) -> Any:
    """Read an uncoerced WAPI collection or its schema through SDK transport.

    This narrow fallback accepts only a single WAPI object type. Schema and
    paging controls must be supplied through ``params``; URLs, object refs,
    embedded query strings, and credential-bearing parameters are rejected.
    Prefer the typed resource APIs whenever one is available.
    """
    _validate_wapi_object_type(object_type)
    _validate_safe_read_params(params)
    result = await self._http.get(object_type, params=dict(params) if params else None)
    return unwrap_value(result)

async def read_schema(self, object_type: str | None = None) -> Any:
    """Read the WAPI root schema or one collection object's schema.

    Args:
        object_type: A safe WAPI collection object type. When omitted, the
            WAPI version root schema is returned.

    Returns:
        The uncoerced JSON schema response from NIOS.

    Raises:
        ValueError: If ``object_type`` is not a safe collection name.
        NiosError: Or a subclass for transport or WAPI errors.
    """
    if object_type is not None:
        _validate_wapi_object_type(object_type)
    result = await self._http.get(object_type or "/", params={"_schema": "1"})
    return unwrap_value(result)

async def aclose(self) -> None:
    """Close the underlying HTTP client and release any active WAPI session.

    Calls :meth:`~ibx_nios_sdk._http.HttpClient.aclose`, which logs out of
    the WAPI session if one is active before closing the ``httpx.AsyncClient``.

    Example:
        Explicit lifecycle management::

            client = NiosClient(grid_url=..., username=..., password=...)
            try:
                ...
            finally:
                await client.aclose()
    """
    await self._http.aclose()

async def request(
    self,
    method: str,
    path: str,
    *,
    params: dict[str, str] | None = None,
    json: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
    """Send a WAPI request with lazy login, retry, and one-shot 401 re-login.

    If ``use_session`` is True and no session exists, authenticates first.
    Retries on 429/5xx status codes up to ``max_retries`` times. On a 401
    after a successful login, re-authenticates once and retries.

    Args:
        method: HTTP method string (e.g. ``"GET"``, ``"POST"``).
        path: WAPI endpoint path relative to ``/wapi/v{version}/``. A
            leading slash is added automatically if absent.
        params: Optional query-string parameters.
        json: Optional JSON request body.

    Returns:
        The parsed JSON response body as a dict, or None for 204 responses.

    Raises:
        NiosConnectionError: If the connection fails before a response is
            received.
        NiosError: Or a subclass for non-success status codes.
    """
    if self._use_session and not self._logged_in:
        await self._login()

    url = path if path.startswith("/") else f"/{path}"
    response = await self._request_with_retry(method, url, params=params, json=json)

    if (
        self._use_session
        and response.status_code == 401
        and self._logged_in
        and not url.endswith("/grid/session")
    ):
        logger.debug("401 on %s - re-logging in", url)
        self._logged_in = False
        await self._login()
        response = await self._request_with_retry(method, url, params=params, json=json)

    return self._handle_response(response)

async def get(
    self, path: str, *, params: dict[str, str] | None = None
) -> dict[str, Any] | None:
    """Send a GET request.

    Args:
        path: WAPI endpoint path relative to the base URL.
        params: Optional query-string parameters.

    Returns:
        The parsed JSON response body, or None for empty responses.

    Raises:
        NiosConnectionError: If the connection fails before a response.
        NiosError: Or a subclass for non-success status codes.
    """
    return await self.request("GET", path, params=params)

async def post(
    self,
    path: str,
    *,
    params: dict[str, str] | None = None,
    json: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
    """Send a POST request.

    Args:
        path: WAPI endpoint path relative to the base URL.
        params: Optional query-string parameters (e.g. ``_function``).
        json: Optional JSON request body.

    Returns:
        The parsed JSON response body, or None for empty responses.

    Raises:
        NiosConnectionError: If the connection fails before a response.
        NiosError: Or a subclass for non-success status codes.
    """
    return await self.request("POST", path, params=params, json=json)

async def put(
    self,
    path: str,
    *,
    params: dict[str, str] | None = None,
    json: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
    """Send a PUT request.

    Args:
        path: WAPI endpoint path relative to the base URL.
        params: Optional query-string parameters (e.g. ``_return_fields``).
        json: Optional JSON request body with the full resource
            representation.

    Returns:
        The parsed JSON response body, or None for empty responses.

    Raises:
        NiosConnectionError: If the connection fails before a response.
        NiosError: Or a subclass for non-success status codes.
    """
    return await self.request("PUT", path, params=params, json=json)

async def delete(
    self, path: str, *, params: dict[str, str] | None = None
) -> dict[str, Any] | None:
    """Send a DELETE request.

    WAPI DELETE responses return the deleted object's ``_ref`` as a plain
    string, which ``_safe_json`` wraps as ``{"_value": ref}``.

    Args:
        path: WAPI endpoint path relative to the base URL.
        params: Optional query-string parameters.

    Returns:
        The parsed JSON response body (typically ``{"_value": ref}``), or
        None for empty responses.

    Raises:
        NiosConnectionError: If the connection fails before a response.
        NiosError: Or a subclass for non-success status codes.
    """
    return await self.request("DELETE", path, params=params)

async def get_with_return_fields(
    self,
    path: str,
    *,
    model_fields: list[str],
    return_fields: list[str] | None = None,
    return_fields_plus: list[str] | None = None,
    extra_params: dict[str, str] | None = None,
) -> dict[str, Any] | None:
    """Send a GET request with ``_return_fields`` / ``_return_fields+`` merged in.

    Delegates field selection logic to
    :func:`~ibx_nios_sdk._query.build_return_fields_params`.

    Args:
        path: WAPI endpoint path relative to the base URL.
        model_fields: Fields defined on the SDK model - used as the default
            set when neither ``return_fields`` nor ``return_fields_plus`` is
            provided.
        return_fields: Exact list of fields to request; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Additional fields to append to ``model_fields``;
            mutually exclusive with ``return_fields``.
        extra_params: Additional query-string parameters merged after field
            params (e.g. ``{"_return_as_object": "1"}``).

    Returns:
        The parsed JSON response body, or None for empty responses.

    Raises:
        ValueError: If both ``return_fields`` and ``return_fields_plus``
            are provided.
        NiosConnectionError: If the connection fails before a response.
        NiosError: Or a subclass for non-success status codes.
    """
    from ibx_nios_sdk._query import build_return_fields_params

    params = build_return_fields_params(
        model_fields=model_fields,
        return_fields=return_fields,
        return_fields_plus=return_fields_plus,
    )
    if extra_params:
        params.update(extra_params)
    return await self.request("GET", path, params=params or None)

async def aclose(self) -> None:
    """Close the client, logging out of any active WAPI session first.

    If a session cookie is active, calls ``_logout`` to invalidate it
    server-side before closing the underlying ``httpx.AsyncClient``.

    Example:
        Explicit close::

            client = HttpClient(grid_url=..., username=..., password=...)
            try:
                ...
            finally:
                await client.aclose()
    """
    if self._use_session and self._logged_in:
        await self._logout()
    await self._client.aclose()

async def list_page(
    self,
    *,
    page_id: str | None = None,
    max_results: int = 1000,
    return_fields: list[str] | None = None,
    return_fields_plus: list[str] | None = None,
    **filters: Any,
) -> tuple[list[TModel], str | None]:
    """Fetch a single page of WAPI results.

    Args:
        page_id: Continuation token from a previous response. Pass ``None``
            to start from the first page.
        max_results: Maximum number of records per page (WAPI
            ``_max_results``).
        return_fields: Exact list of fields to return; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Extra fields to add on top of
            ``_default_return_fields``; mutually exclusive with
            ``return_fields``.
        **filters: Field-level filters using kwarg conventions (e.g.
            ``name="foo"``, ``comment__like="bar"``,
            ``extattr_Site="NYC"``).

    Returns:
        A ``(rows, next_page_id)`` tuple. ``rows`` is a list of validated
        model instances; ``next_page_id`` is the token for the next page, or
        ``None`` if this was the last page.

    Raises:
        ValueError: If both ``return_fields`` and ``return_fields_plus`` are
            provided, or if an unknown filter operator suffix is used.
        UnsupportedOperationError: If WAPI does not support reads of this
            object type.
        NiosError: Or a subclass for WAPI errors.

    Example:
        Fetch the first two pages manually::

            rows, next_id = await resource.list_page(max_results=100)
            if next_id:
                more, _ = await resource.list_page(page_id=next_id)
    """
    self._require_op("read")
    params: dict[str, str] = {
        "_paging": "1",
        "_return_as_object": "1",
        "_max_results": str(max_results),
    }
    params.update(build_filter_params(filters))
    params.update(
        build_return_fields_params(
            model_fields=self._default_return_fields,
            return_fields=return_fields,
            return_fields_plus=return_fields_plus,
        )
    )
    if page_id:
        params["_page_id"] = page_id

    data = await self._client.get(self._path(), params=params)
    body = data or {}
    rows = body.get("result", []) or []
    next_id = body.get("next_page_id") or None
    parsed = [self._model.model_validate(row) for row in rows]
    return parsed, next_id  # type: ignore[return-value]

def list(
    self,
    *,
    max_results: int = 1000,
    return_fields: list[str] | None = None,
    return_fields_plus: list[str] | None = None,
    **filters: Any,
) -> AsyncPageIterator[TModel]:
    """Return an auto-paginated async iterator over all matching objects.

    Fetches pages on demand as the caller iterates. Use ``await .all()``
    on the returned iterator to collect every record into a list at once.

    Args:
        max_results: Maximum records per page (WAPI ``_max_results``).
        return_fields: Exact list of fields to return; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Extra fields on top of ``_default_return_fields``;
            mutually exclusive with ``return_fields``.
        **filters: Field-level filters (see :func:`build_filter_params`).

    Returns:
        An :class:`~ibx_nios_sdk._paging.AsyncPageIterator` that yields
        validated model instances across all pages.

    Raises:
        UnsupportedOperationError: If WAPI does not support reads of this
            object type. Raised by this call, not by the first iteration.

    Example:
        Iterate lazily::

            async for record in resource.list(name="example"):
                print(record.ref)

        Collect all at once::

            records = await resource.list(comment__like="prod").all()
    """

    self._require_op("read")

    async def _fetch(page_id: str | None) -> tuple[list[dict[str, Any]], str | None]:
        params: dict[str, str] = {
            "_paging": "1",
            "_return_as_object": "1",
            "_max_results": str(max_results),
        }
        params.update(build_filter_params(filters))
        params.update(
            build_return_fields_params(
                model_fields=self._default_return_fields,
                return_fields=return_fields,
                return_fields_plus=return_fields_plus,
            )
        )
        if page_id:
            params["_page_id"] = page_id
        data = await self._client.get(self._path(), params=params)
        body = data or {}
        return body.get("result", []) or [], body.get("next_page_id") or None

    return AsyncPageIterator(fetch=_fetch, model=self._model)  # type: ignore[arg-type]

async def find_one(
    self,
    *,
    return_fields: _StrList | None = None,
    return_fields_plus: _StrList | None = None,
    **filters: Any,
) -> TModel | None:
    """Return the first matching object, or None if no match is found.

    Requests at most one result (``max_results=1``) to minimise data
    transfer.

    Args:
        return_fields: Exact list of fields to return; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Extra fields on top of ``_default_return_fields``;
            mutually exclusive with ``return_fields``.
        **filters: Field-level filters (see :func:`build_filter_params`).

    Returns:
        The first matching validated model instance, or ``None`` if the
        WAPI returns an empty result set.

    Raises:
        NiosError: Or a subclass for WAPI errors.

    Example:
        Look up a record by name::

            record = await resource.find_one(name="foo.example.com")
            if record is None:
                print("not found")
    """
    rows, _ = await self.list_page(
        max_results=1,
        return_fields=return_fields,
        return_fields_plus=return_fields_plus,
        **filters,
    )
    return rows[0] if rows else None

async def get(
    self,
    ref: str,
    *,
    return_fields: _StrList | None = None,
    return_fields_plus: _StrList | None = None,
) -> TModel:
    """Fetch a single WAPI object by its ``_ref``.

    Args:
        ref: The WAPI object reference string (e.g.
            ``"record:a/ZG5z...:1.2.3.4/default"``).
        return_fields: Exact list of fields to return; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Extra fields on top of ``_default_return_fields``;
            mutually exclusive with ``return_fields``.

    Returns:
        A validated model instance for the requested object.

    Raises:
        ValueError: If ``ref`` does not belong to this resource type.
        NotFoundError: If the object no longer exists in NIOS.
        UnsupportedOperationError: If WAPI does not support reads of this
            object type.
        NiosError: Or a subclass for other WAPI errors.

    Example:
        Fetch a record by ref::

            record = await resource.get("record:a/ZG5z...:1.2.3.4/default")
            print(record.ipv4addr)
    """
    self._require_op("read")
    self._validate_ref(ref)
    params = build_return_fields_params(
        model_fields=self._default_return_fields,
        return_fields=return_fields,
        return_fields_plus=return_fields_plus,
    )
    data = await self._client.get(f"/{ref}", params=params or None)
    return self._model.model_validate(data or {})  # type: ignore[return-value]

async def create(
    self,
    obj: BaseModel | dict[str, Any],
    *,
    return_fields: _StrList | None = None,
    return_fields_plus: _StrList | None = None,
) -> TModel:
    """Create a new WAPI object and return the populated model.

    Args:
        obj: A Pydantic model instance or plain dict containing the fields
            for the new object. ``None`` values and the ``ref`` field are
            excluded from the POST body automatically.
        return_fields: Exact list of fields to return; mutually exclusive
            with ``return_fields_plus``.
        return_fields_plus: Extra fields on top of ``_default_return_fields``;
            mutually exclusive with ``return_fields``.

    Returns:
        A validated model instance for the newly created object, populated
        with the fields requested via ``_return_as_object=1``.

    Raises:
        UnsupportedOperationError: If WAPI does not support creating this
            object type.
        ConflictError: If the object already exists.
        BadRequestError: If the request body contains invalid fields.
        NiosError: Or a subclass for other WAPI errors.

    Example:
        Create an A record::

            record = await resource.create(
                {"name": "host.example.com", "ipv4addr": "1.2.3.4"}
            )
            print(record.ref)
    """
    self._require_op("create")
    payload = self._dump(obj, exclude_readonly=False)
    params: dict[str, str] = {"_return_as_object": "1"}
    params.update(
        build_return_fields_params(
            model_fields=self._default_return_fields,
            return_fields=return_fields,
            return_fields_plus=return_fields_plus,
        )
    )
    data = await self._client.post(self._path(), params=params, json=payload)
    if isinstance(data, dict) and list(data) == ["result"]:
        data = data["result"]
    return self._model.model_validate(data or {})  # type: ignore[return-value]