GlpiOAuthClient

OAuth2 Clients Clients for the GLPI 11 high-level API (/api.php).

GlpiOAuthClient

glpi_utils.oauth.GlpiOAuthClient ¶

Synchronous OAuth2 client for the GLPI 11 High-level API (/api.php).

Parameters:

Name	Type	Description	Default
`url`	`str or None`	Base URL, e.g. `"https://glpi.example.com"`. Falls back to `GLPI_URL`.	`None`
`client_id`	`str or None`	OAuth2 client ID. Falls back to `GLPI_OAUTH_CLIENT_ID`.	`None`
`client_secret`	`str or None`	OAuth2 client secret. Falls back to `GLPI_OAUTH_CLIENT_SECRET`.	`None`
`verify_ssl`	`bool`	Verify TLS certificates (default: `True`).	`True`
`timeout`	`int`	HTTP timeout in seconds (default: `30`).	`30`

Source code in glpi_utils/oauth.py

class GlpiOAuthClient:
    """Synchronous OAuth2 client for the GLPI 11 High-level API (``/api.php``).

    Parameters
    ----------
    url : str or None
        Base URL, e.g. ``"https://glpi.example.com"``.
        Falls back to ``GLPI_URL``.
    client_id : str or None
        OAuth2 client ID. Falls back to ``GLPI_OAUTH_CLIENT_ID``.
    client_secret : str or None
        OAuth2 client secret. Falls back to ``GLPI_OAUTH_CLIENT_SECRET``.
    verify_ssl : bool
        Verify TLS certificates (default: ``True``).
    timeout : int
        HTTP timeout in seconds (default: ``30``).
    """

    def __init__(
        self,
        url: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
        verify_ssl: bool = True,
        timeout: int = 30,
    ) -> None:
        self._url = (url or os.environ.get("GLPI_URL", "")).rstrip("/")
        if not self._url:
            raise ValueError("A GLPI URL is required. Pass url= or set GLPI_URL.")

        self._client_id = client_id or os.environ.get("GLPI_OAUTH_CLIENT_ID")
        self._client_secret = client_secret or os.environ.get("GLPI_OAUTH_CLIENT_SECRET")
        self._username = username or os.environ.get("GLPI_OAUTH_USERNAME")
        self._password = password or os.environ.get("GLPI_OAUTH_PASSWORD")
        self._verify_ssl = verify_ssl
        self._timeout = timeout

        self._token = _TokenStore()
        self._http = Session()
        self._version: Optional[GLPIVersion] = None
        self._proxies: dict = {}

    # ------------------------------------------------------------------
    # Context manager
    # ------------------------------------------------------------------

    def __enter__(self) -> "GlpiOAuthClient":
        return self

    def __exit__(self, *_: Any) -> None:
        self.close()

    def close(self) -> None:
        """Close the underlying HTTP session."""
        self._http.close()
        self._token.clear()

    # ------------------------------------------------------------------
    # Fluent accessors
    # ------------------------------------------------------------------

    def __getattr__(self, name: str) -> ItemProxy:
        # Use object.__getattribute__ to avoid recursive __getattr__ calls
        # when _proxies itself hasn't been set yet.
        try:
            proxies = object.__getattribute__(self, "_proxies")
        except AttributeError:
            proxies = {}
            object.__setattr__(self, "_proxies", proxies)

        lower = name.lower()
        if lower in _ITEMTYPE_MAP:
            if lower not in proxies:
                proxies[lower] = ItemProxy(self, _ITEMTYPE_MAP[lower])  # type: ignore[arg-type]
            return proxies[lower]
        raise AttributeError(
            f"{self.__class__.__name__!r} has no attribute {name!r}. "
            "Use api.item('YourItemtype') for non-standard item types."
        )

    def item(self, itemtype: str) -> ItemProxy:
        """Return an :class:`~glpi_utils._resource.ItemProxy` for any itemtype."""
        try:
            proxies = object.__getattribute__(self, "_proxies")
        except AttributeError:
            proxies = {}
            object.__setattr__(self, "_proxies", proxies)
        if itemtype not in proxies:
            proxies[itemtype] = ItemProxy(self, itemtype)  # type: ignore[arg-type]
        return proxies[itemtype]

    # ------------------------------------------------------------------
    # URL helpers
    # ------------------------------------------------------------------

    @property
    def _token_url(self) -> str:
        return f"{self._url}/api.php/token"

    @property
    def _api_url(self) -> str:
        return f"{self._url}/api.php"

    def _auth_headers(self, with_content_type: bool = False) -> dict:
        token = self._token.access_token
        if not token:
            raise GlpiAuthError("Not authenticated. Call authenticate() first.")
        headers = {"Authorization": f"Bearer {token}"}
        if with_content_type:
            headers["Content-Type"] = "application/json"
        return headers

    # ------------------------------------------------------------------
    # Authentication
    # ------------------------------------------------------------------

    def authenticate(
        self,
        username: Optional[str] = None,
        password: Optional[str] = None,
    ) -> None:
        """Obtain an OAuth2 Bearer token.

        Parameters
        ----------
        username : str or None
            If provided (along with *password*), uses the ``password`` grant.
            Otherwise uses ``client_credentials``.
            Falls back to constructor username or ``GLPI_OAUTH_USERNAME``.
        password : str or None
        """
        if not self._client_id:
            raise GlpiAuthError(
                "client_id is required. Pass it or set GLPI_OAUTH_CLIENT_ID."
            )

        username = username or self._username
        password = password or self._password

        if username and password:
            payload = {
                "grant_type": "password",
                "client_id": self._client_id,
                "client_secret": self._client_secret or "",
                "username": username,
                "password": password,
                "scope": "api",
            }
            grant = "password"
        else:
            if not self._client_secret:
                raise GlpiAuthError(
                    "client_secret is required for client_credentials grant. "
                    "Set GLPI_OAUTH_CLIENT_SECRET or pass client_secret=."
                )
            payload = {
                "grant_type": "client_credentials",
                "client_id": self._client_id,
                "client_secret": self._client_secret,
                "scope": "api",
            }
            grant = "client_credentials"

        log.debug("OAuth2 token request (grant=%s)", grant)

        try:
            resp = self._http.post(
                self._token_url,
                data=payload,
                verify=self._verify_ssl,
                timeout=self._timeout,
            )
        except requests.exceptions.ConnectionError as exc:
            raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc
        except requests.exceptions.Timeout as exc:
            raise GlpiConnectionError(f"Token request timed out: {exc}") from exc

        body = resp.json()
        if resp.status_code not in (200, 201) or "access_token" not in body:
            error = body.get("error", str(resp.status_code))
            desc  = body.get("error_description", resp.text)
            raise GlpiAuthError(
                f"OAuth2 token request failed ({error}): {desc}",
                status_code=resp.status_code,
                error_code=error,
            )

        self._token.store(body["access_token"], int(body.get("expires_in", 3600)))
        log.debug("OAuth2 token obtained (expires_in=%s).", body.get("expires_in"))

    def _ensure_token(
        self,
        username: Optional[str] = None,
        password: Optional[str] = None,
    ) -> None:
        """Re-authenticate automatically if the token has expired."""
        if not self._token.is_valid():
            log.debug("Token expired or missing — re-authenticating.")
            self.authenticate(username=username, password=password)

    # ------------------------------------------------------------------
    # Internal HTTP helpers
    # ------------------------------------------------------------------

    def _request(
        self,
        method: str,
        path: str,
        *,
        params: Optional[dict] = None,
        json: Any = None,
    ) -> Any:
        body, _ = self._request_with_headers(method, path, params=params, json=json)
        return body

    def _request_with_headers(
        self,
        method: str,
        path: str,
        *,
        params: Optional[dict] = None,
        json: Any = None,
    ) -> tuple:
        self._ensure_token()
        url = f"{self._api_url}/{path.lstrip('/')}"
        log.debug("%s %s params=%s", method.upper(), url, params)

        try:
            resp = self._http.request(
                method, url,
                headers=self._auth_headers(with_content_type=json is not None),
                params=params,
                json=json,
                verify=self._verify_ssl,
                timeout=self._timeout,
            )
        except requests.exceptions.ConnectionError as exc:
            raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc
        except requests.exceptions.Timeout as exc:
            raise GlpiConnectionError(f"Request timed out: {exc}") from exc

        log.debug("Response %s from %s", resp.status_code, url)

        if resp.status_code == 204 or not resp.content:
            return None, dict(resp.headers)

        try:
            body = resp.json()
        except Exception:
            body = resp.text

        _raise_for_oauth_error(resp.status_code, body, resp.text)
        return body, dict(resp.headers)

    # ------------------------------------------------------------------
    # Version
    # ------------------------------------------------------------------

    @property
    def version(self) -> Optional[GLPIVersion]:
        """Return cached server version."""
        return self._version

    def get_version(self) -> GLPIVersion:
        """Fetch and cache the GLPI server version."""
        data = self._request("GET", "")
        # High-level API root returns server info
        raw = data.get("glpi_version") if isinstance(data, dict) else None
        self._version = GLPIVersion(raw or "0.0.0")
        return self._version

    # ------------------------------------------------------------------
    # Item CRUD
    # ------------------------------------------------------------------

    def get_item(self, itemtype: str, item_id: int, **kwargs: Any) -> dict:
        """Return a single item by ID."""
        params = _boolify_params(kwargs)
        return self._request("GET", f"{_hl_route(itemtype)}/{item_id}", params=params)

    def get_all_items(self, itemtype: str, **kwargs: Any) -> list:
        """Return a single page of items."""
        params = _boolify_params(kwargs)
        if "range" not in params:
            params["range"] = f"0-{DEFAULT_PAGE_SIZE - 1}"
        return self._request("GET", _hl_route(itemtype), params=params)

    def get_all_pages(
        self,
        itemtype: str,
        page_size: int = DEFAULT_PAGE_SIZE,
        **kwargs: Any,
    ) -> list:
        """Fetch all items across all pages automatically."""
        return list(item for page in self.iter_pages(itemtype, page_size=page_size, **kwargs) for item in page)

    def iter_pages(
        self,
        itemtype: str,
        page_size: int = DEFAULT_PAGE_SIZE,
        **kwargs: Any,
    ) -> Iterator[list]:
        """Yield one page of items at a time."""
        params = _boolify_params(kwargs)
        start = 0
        fetched = 0

        while True:
            end = start + page_size - 1
            params["range"] = f"{start}-{end}"

            page_items, resp_headers = self._request_with_headers("GET", _hl_route(itemtype), params=params)

            if not page_items:
                return

            fetched += len(page_items)
            yield page_items

            total = _parse_content_range(resp_headers.get("Content-Range", ""))
            if total is not None and fetched >= total:
                return
            if len(page_items) < page_size:
                return

            start += page_size

    def search(self, itemtype: str, **kwargs: Any) -> dict:
        """Run the GLPI search engine."""
        params = _boolify_params(kwargs)
        return self._request("GET", f"search/{_hl_route(itemtype)}", params=params)

    def create_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> Any:
        """Create one or several items."""
        # HL API v2: POST body is the object directly, no {input:} wrapper
        if isinstance(input_data, dict):
            body = dict(input_data)
            body.update(kwargs)
        else:
            body = input_data
        return self._request("POST", _hl_route(itemtype), json=body)

    def update_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> list:
        """Update one or several items."""
        # HL API v2: PUT /{route}/{id} with fields in body (no wrapper)
        if isinstance(input_data, dict) and "id" in input_data:
            item_id = input_data["id"]
            body = {k: v for k, v in input_data.items() if k != "id"}
            body.update(kwargs)
            return self._request("PATCH", f"{_hl_route(itemtype)}/{item_id}", json=body)
        # Fallback for list input (bulk) - not natively supported in HL API
        payload: dict = {"input": input_data}
        payload.update(kwargs)
        return self._request("PATCH", _hl_route(itemtype), json=payload)

    def delete_item(
        self,
        itemtype: str,
        input_data: Any,
        force_purge: bool = False,
        history: bool = True,
    ) -> list:
        """Delete one or several items."""
        # HL API v2: DELETE /{route}/{id}
        if isinstance(input_data, dict) and "id" in input_data:
            item_id = input_data["id"]
            return self._request("DELETE", f"{_hl_route(itemtype)}/{item_id}")
        payload: dict = {"input": input_data}
        return self._request("DELETE", _hl_route(itemtype), json=payload)

    def get_sub_items(
        self, itemtype: str, item_id: int, sub_itemtype: str, **kwargs: Any
    ) -> list:
        """Return sub-items of a parent item."""
        params = _boolify_params(kwargs)
        return self._request("GET", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", params=params)

    def add_sub_item(
        self, itemtype: str, item_id: int, sub_itemtype: str, input_data: dict, **kwargs: Any
    ) -> dict:
        """Add a sub-item to a parent resource."""
        # HL API v2: body is the object directly, no {"input":...} wrapper
        body = dict(input_data)
        body.update(kwargs)
        return self._request("POST", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", json=body)

version `property` ¶

version: Optional[GLPIVersion]

Return cached server version.

close ¶

close() -> None

Close the underlying HTTP session.

Source code in glpi_utils/oauth.py

def close(self) -> None:
    """Close the underlying HTTP session."""
    self._http.close()
    self._token.clear()

item ¶

item(itemtype: str) -> ItemProxy

Return an :class:~glpi_utils._resource.ItemProxy for any itemtype.

Source code in glpi_utils/oauth.py

def item(self, itemtype: str) -> ItemProxy:
    """Return an :class:`~glpi_utils._resource.ItemProxy` for any itemtype."""
    try:
        proxies = object.__getattribute__(self, "_proxies")
    except AttributeError:
        proxies = {}
        object.__setattr__(self, "_proxies", proxies)
    if itemtype not in proxies:
        proxies[itemtype] = ItemProxy(self, itemtype)  # type: ignore[arg-type]
    return proxies[itemtype]

authenticate ¶

authenticate(username: Optional[str] = None, password: Optional[str] = None) -> None

Obtain an OAuth2 Bearer token.

Parameters:

Name	Type	Description	Default
`username`	`str or None`	If provided (along with password), uses the `password` grant. Otherwise uses `client_credentials`. Falls back to constructor username or `GLPI_OAUTH_USERNAME`.	`None`
`password`	`str or None`		`None`

Source code in glpi_utils/oauth.py

def authenticate(
    self,
    username: Optional[str] = None,
    password: Optional[str] = None,
) -> None:
    """Obtain an OAuth2 Bearer token.

    Parameters
    ----------
    username : str or None
        If provided (along with *password*), uses the ``password`` grant.
        Otherwise uses ``client_credentials``.
        Falls back to constructor username or ``GLPI_OAUTH_USERNAME``.
    password : str or None
    """
    if not self._client_id:
        raise GlpiAuthError(
            "client_id is required. Pass it or set GLPI_OAUTH_CLIENT_ID."
        )

    username = username or self._username
    password = password or self._password

    if username and password:
        payload = {
            "grant_type": "password",
            "client_id": self._client_id,
            "client_secret": self._client_secret or "",
            "username": username,
            "password": password,
            "scope": "api",
        }
        grant = "password"
    else:
        if not self._client_secret:
            raise GlpiAuthError(
                "client_secret is required for client_credentials grant. "
                "Set GLPI_OAUTH_CLIENT_SECRET or pass client_secret=."
            )
        payload = {
            "grant_type": "client_credentials",
            "client_id": self._client_id,
            "client_secret": self._client_secret,
            "scope": "api",
        }
        grant = "client_credentials"

    log.debug("OAuth2 token request (grant=%s)", grant)

    try:
        resp = self._http.post(
            self._token_url,
            data=payload,
            verify=self._verify_ssl,
            timeout=self._timeout,
        )
    except requests.exceptions.ConnectionError as exc:
        raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc
    except requests.exceptions.Timeout as exc:
        raise GlpiConnectionError(f"Token request timed out: {exc}") from exc

    body = resp.json()
    if resp.status_code not in (200, 201) or "access_token" not in body:
        error = body.get("error", str(resp.status_code))
        desc  = body.get("error_description", resp.text)
        raise GlpiAuthError(
            f"OAuth2 token request failed ({error}): {desc}",
            status_code=resp.status_code,
            error_code=error,
        )

    self._token.store(body["access_token"], int(body.get("expires_in", 3600)))
    log.debug("OAuth2 token obtained (expires_in=%s).", body.get("expires_in"))

get_version ¶

get_version() -> GLPIVersion

Fetch and cache the GLPI server version.

Source code in glpi_utils/oauth.py

def get_version(self) -> GLPIVersion:
    """Fetch and cache the GLPI server version."""
    data = self._request("GET", "")
    # High-level API root returns server info
    raw = data.get("glpi_version") if isinstance(data, dict) else None
    self._version = GLPIVersion(raw or "0.0.0")
    return self._version

get_item ¶

get_item(itemtype: str, item_id: int, **kwargs: Any) -> dict

Return a single item by ID.

Source code in glpi_utils/oauth.py

def get_item(self, itemtype: str, item_id: int, **kwargs: Any) -> dict:
    """Return a single item by ID."""
    params = _boolify_params(kwargs)
    return self._request("GET", f"{_hl_route(itemtype)}/{item_id}", params=params)

get_all_items ¶

get_all_items(itemtype: str, **kwargs: Any) -> list

Return a single page of items.

Source code in glpi_utils/oauth.py

def get_all_items(self, itemtype: str, **kwargs: Any) -> list:
    """Return a single page of items."""
    params = _boolify_params(kwargs)
    if "range" not in params:
        params["range"] = f"0-{DEFAULT_PAGE_SIZE - 1}"
    return self._request("GET", _hl_route(itemtype), params=params)

get_all_pages ¶

get_all_pages(itemtype: str, page_size: int = DEFAULT_PAGE_SIZE, **kwargs: Any) -> list

Fetch all items across all pages automatically.

Source code in glpi_utils/oauth.py

def get_all_pages(
    self,
    itemtype: str,
    page_size: int = DEFAULT_PAGE_SIZE,
    **kwargs: Any,
) -> list:
    """Fetch all items across all pages automatically."""
    return list(item for page in self.iter_pages(itemtype, page_size=page_size, **kwargs) for item in page)

iter_pages ¶

iter_pages(itemtype: str, page_size: int = DEFAULT_PAGE_SIZE, **kwargs: Any) -> Iterator[list]

Yield one page of items at a time.

Source code in glpi_utils/oauth.py

def iter_pages(
    self,
    itemtype: str,
    page_size: int = DEFAULT_PAGE_SIZE,
    **kwargs: Any,
) -> Iterator[list]:
    """Yield one page of items at a time."""
    params = _boolify_params(kwargs)
    start = 0
    fetched = 0

    while True:
        end = start + page_size - 1
        params["range"] = f"{start}-{end}"

        page_items, resp_headers = self._request_with_headers("GET", _hl_route(itemtype), params=params)

        if not page_items:
            return

        fetched += len(page_items)
        yield page_items

        total = _parse_content_range(resp_headers.get("Content-Range", ""))
        if total is not None and fetched >= total:
            return
        if len(page_items) < page_size:
            return

        start += page_size

search ¶

search(itemtype: str, **kwargs: Any) -> dict

Run the GLPI search engine.

Source code in glpi_utils/oauth.py

def search(self, itemtype: str, **kwargs: Any) -> dict:
    """Run the GLPI search engine."""
    params = _boolify_params(kwargs)
    return self._request("GET", f"search/{_hl_route(itemtype)}", params=params)

create_item ¶

create_item(itemtype: str, input_data: Any, **kwargs: Any) -> Any

Create one or several items.

Source code in glpi_utils/oauth.py

def create_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> Any:
    """Create one or several items."""
    # HL API v2: POST body is the object directly, no {input:} wrapper
    if isinstance(input_data, dict):
        body = dict(input_data)
        body.update(kwargs)
    else:
        body = input_data
    return self._request("POST", _hl_route(itemtype), json=body)

update_item ¶

update_item(itemtype: str, input_data: Any, **kwargs: Any) -> list

Update one or several items.

Source code in glpi_utils/oauth.py

def update_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> list:
    """Update one or several items."""
    # HL API v2: PUT /{route}/{id} with fields in body (no wrapper)
    if isinstance(input_data, dict) and "id" in input_data:
        item_id = input_data["id"]
        body = {k: v for k, v in input_data.items() if k != "id"}
        body.update(kwargs)
        return self._request("PATCH", f"{_hl_route(itemtype)}/{item_id}", json=body)
    # Fallback for list input (bulk) - not natively supported in HL API
    payload: dict = {"input": input_data}
    payload.update(kwargs)
    return self._request("PATCH", _hl_route(itemtype), json=payload)

delete_item ¶

delete_item(itemtype: str, input_data: Any, force_purge: bool = False, history: bool = True) -> list

Delete one or several items.

Source code in glpi_utils/oauth.py

def delete_item(
    self,
    itemtype: str,
    input_data: Any,
    force_purge: bool = False,
    history: bool = True,
) -> list:
    """Delete one or several items."""
    # HL API v2: DELETE /{route}/{id}
    if isinstance(input_data, dict) and "id" in input_data:
        item_id = input_data["id"]
        return self._request("DELETE", f"{_hl_route(itemtype)}/{item_id}")
    payload: dict = {"input": input_data}
    return self._request("DELETE", _hl_route(itemtype), json=payload)

get_sub_items ¶

get_sub_items(itemtype: str, item_id: int, sub_itemtype: str, **kwargs: Any) -> list

Return sub-items of a parent item.

Source code in glpi_utils/oauth.py

def get_sub_items(
    self, itemtype: str, item_id: int, sub_itemtype: str, **kwargs: Any
) -> list:
    """Return sub-items of a parent item."""
    params = _boolify_params(kwargs)
    return self._request("GET", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", params=params)

add_sub_item ¶

add_sub_item(itemtype: str, item_id: int, sub_itemtype: str, input_data: dict, **kwargs: Any) -> dict

Add a sub-item to a parent resource.

Source code in glpi_utils/oauth.py

def add_sub_item(
    self, itemtype: str, item_id: int, sub_itemtype: str, input_data: dict, **kwargs: Any
) -> dict:
    """Add a sub-item to a parent resource."""
    # HL API v2: body is the object directly, no {"input":...} wrapper
    body = dict(input_data)
    body.update(kwargs)
    return self._request("POST", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", json=body)

AsyncGlpiOAuthClient

glpi_utils.oauth.AsyncGlpiOAuthClient ¶

Asynchronous OAuth2 client for the GLPI 11 High-level API (/api.php).

Parameters:

Name	Type	Default
`url`	`str or None`	`None`
`client_id`	`str or None`	`None`
`client_secret`	`str or None`	`None`
`verify_ssl`	`bool`	`True`
`timeout`	`int`	`30`

Examples:

::

async with AsyncGlpiOAuthClient(
    url="https://glpi.example.com",
    client_id="my-app",
    client_secret="secret",
) as api:
    await api.authenticate()
    tickets = await api.ticket.get_all_pages()

Source code in glpi_utils/oauth.py

class AsyncGlpiOAuthClient:
    """Asynchronous OAuth2 client for the GLPI 11 High-level API (``/api.php``).

    Parameters
    ----------
    url : str or None
    client_id : str or None
    client_secret : str or None
    verify_ssl : bool
    timeout : int

    Examples
    --------
    ::

        async with AsyncGlpiOAuthClient(
            url="https://glpi.example.com",
            client_id="my-app",
            client_secret="secret",
        ) as api:
            await api.authenticate()
            tickets = await api.ticket.get_all_pages()
    """

    def __init__(
        self,
        url: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
        verify_ssl: bool = True,
        timeout: int = 30,
    ) -> None:
        try:
            import aiohttp  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                "AsyncGlpiOAuthClient requires 'aiohttp'. "
                "Install it with: pip install glpi-utils[async]"
            ) from exc

        self._url = (url or os.environ.get("GLPI_URL", "")).rstrip("/")
        if not self._url:
            raise ValueError("A GLPI URL is required. Pass url= or set GLPI_URL.")

        self._client_id = client_id or os.environ.get("GLPI_OAUTH_CLIENT_ID")
        self._client_secret = client_secret or os.environ.get("GLPI_OAUTH_CLIENT_SECRET")
        self._username = username or os.environ.get("GLPI_OAUTH_USERNAME")
        self._password = password or os.environ.get("GLPI_OAUTH_PASSWORD")
        self._verify_ssl = verify_ssl
        self._timeout = timeout

        self._token = _TokenStore()
        self._http: Any = None
        self._version: Optional[GLPIVersion] = None
        self._proxies: dict = {}

    async def __aenter__(self) -> "AsyncGlpiOAuthClient":
        return self

    async def __aexit__(self, *_: Any) -> None:
        await self.close()

    async def close(self) -> None:
        """Close the underlying aiohttp session."""
        if self._http and not self._http.closed:
            await self._http.close()
        self._token.clear()

    # ------------------------------------------------------------------
    # Fluent accessors
    # ------------------------------------------------------------------

    def __getattr__(self, name: str) -> AsyncItemProxy:
        try:
            proxies = object.__getattribute__(self, "_proxies")
        except AttributeError:
            proxies = {}
            object.__setattr__(self, "_proxies", proxies)

        lower = name.lower()
        if lower in _ITEMTYPE_MAP:
            if lower not in proxies:
                proxies[lower] = AsyncItemProxy(self, _ITEMTYPE_MAP[lower])  # type: ignore[arg-type]
            return proxies[lower]
        raise AttributeError(
            f"{self.__class__.__name__!r} has no attribute {name!r}. "
            "Use api.item('YourItemtype') for non-standard item types."
        )

    def item(self, itemtype: str) -> AsyncItemProxy:
        try:
            proxies = object.__getattribute__(self, "_proxies")
        except AttributeError:
            proxies = {}
            object.__setattr__(self, "_proxies", proxies)
        if itemtype not in proxies:
            proxies[itemtype] = AsyncItemProxy(self, itemtype)  # type: ignore[arg-type]
        return proxies[itemtype]

    # ------------------------------------------------------------------
    # HTTP helpers
    # ------------------------------------------------------------------

    @property
    def _token_url(self) -> str:
        return f"{self._url}/api.php/token"

    @property
    def _api_url(self) -> str:
        return f"{self._url}/api.php"

    def _get_http(self) -> Any:
        import aiohttp

        if self._http is None or self._http.closed:
            connector = aiohttp.TCPConnector(ssl=self._verify_ssl)
            self._http = aiohttp.ClientSession(connector=connector)
        return self._http

    def _auth_headers(self, with_content_type: bool = False) -> dict:
        if not self._token.access_token:
            raise GlpiAuthError("Not authenticated. Call await authenticate() first.")
        headers = {"Authorization": f"Bearer {self._token.access_token}"}
        if with_content_type:
            headers["Content-Type"] = "application/json"
        return headers

    # ------------------------------------------------------------------
    # Authentication
    # ------------------------------------------------------------------

    async def authenticate(
        self,
        username: Optional[str] = None,
        password: Optional[str] = None,
    ) -> None:
        """Obtain an OAuth2 Bearer token asynchronously."""
        import aiohttp

        if not self._client_id:
            raise GlpiAuthError("client_id is required. Set GLPI_OAUTH_CLIENT_ID.")

        username = username or self._username
        password = password or self._password

        if username and password:
            payload = {
                "grant_type": "password",
                "client_id": self._client_id,
                "client_secret": self._client_secret or "",
                "username": username,
                "password": password,
                "scope": "api",
            }
            grant = "password"
        else:
            if not self._client_secret:
                raise GlpiAuthError(
                    "client_secret is required for client_credentials grant."
                )
            payload = {
                "grant_type": "client_credentials",
                "client_id": self._client_id,
                "client_secret": self._client_secret,
                "scope": "api",
            }
            grant = "client_credentials"

        log.debug("Async OAuth2 token request (grant=%s)", grant)

        http = self._get_http()
        timeout = aiohttp.ClientTimeout(total=self._timeout)

        try:
            async with http.post(
                self._token_url, data=payload, timeout=timeout
            ) as resp:
                body = await resp.json(content_type=None)
                if resp.status not in (200, 201) or "access_token" not in body:
                    error = body.get("error", str(resp.status))
                    desc  = body.get("error_description", "")
                    raise GlpiAuthError(
                        f"OAuth2 token request failed ({error}): {desc}",
                        status_code=resp.status,
                        error_code=error,
                    )
                self._token.store(body["access_token"], int(body.get("expires_in", 3600)))
                log.debug("Async OAuth2 token obtained.")
        except aiohttp.ClientConnectorError as exc:
            raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc

    async def _ensure_token(self) -> None:
        if not self._token.is_valid():
            await self.authenticate()

    # ------------------------------------------------------------------
    # Internal HTTP helpers
    # ------------------------------------------------------------------

    async def _request(self, method: str, path: str, *, params: Optional[dict] = None, json: Any = None) -> Any:
        body, _ = await self._request_with_headers(method, path, params=params, json=json)
        return body

    async def _request_with_headers(
        self, method: str, path: str, *, params: Optional[dict] = None, json: Any = None
    ) -> tuple:
        import aiohttp

        await self._ensure_token()
        url = f"{self._api_url}/{path.lstrip('/')}"
        log.debug("ASYNC %s %s params=%s", method.upper(), url, params)

        http = self._get_http()
        timeout = aiohttp.ClientTimeout(total=self._timeout)

        try:
            async with http.request(
                method, url,
                headers=self._auth_headers(with_content_type=json is not None),
                params=params,
                json=json,
                timeout=timeout,
            ) as resp:
                resp_headers = dict(resp.headers)
                if resp.status == 204 or resp.content_length == 0:
                    return None, resp_headers
                body = await resp.json(content_type=None)
                _raise_for_oauth_error(resp.status, body, "")
                return body, resp_headers
        except aiohttp.ClientConnectorError as exc:
            raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc
        except aiohttp.ServerTimeoutError as exc:
            raise GlpiConnectionError(f"Request timed out: {exc}") from exc

    # ------------------------------------------------------------------
    # CRUD + pagination  (mirrors GlpiOAuthClient)
    # ------------------------------------------------------------------

    async def get_item(self, itemtype: str, item_id: int, **kwargs: Any) -> dict:
        return await self._request("GET", f"{_hl_route(itemtype)}/{item_id}", params=_boolify_params(kwargs))

    async def get_all_items(self, itemtype: str, **kwargs: Any) -> list:
        params = _boolify_params(kwargs)
        if "range" not in params:
            params["range"] = f"0-{DEFAULT_PAGE_SIZE - 1}"
        return await self._request("GET", _hl_route(itemtype), params=params)

    async def get_all_pages(self, itemtype: str, page_size: int = DEFAULT_PAGE_SIZE, **kwargs: Any) -> list:
        results: list = []
        async for page in self.iter_pages(itemtype, page_size=page_size, **kwargs):
            results.extend(page)
        return results

    async def iter_pages(
        self, itemtype: str, page_size: int = DEFAULT_PAGE_SIZE, **kwargs: Any
    ) -> AsyncIterator[list]:
        params = _boolify_params(kwargs)
        start = 0
        fetched = 0
        while True:
            end = start + page_size - 1
            params["range"] = f"{start}-{end}"
            page_items, resp_headers = await self._request_with_headers("GET", _hl_route(itemtype), params=params)
            if not page_items:
                return
            fetched += len(page_items)
            yield page_items
            total = _parse_content_range(resp_headers.get("Content-Range", ""))
            if total is not None and fetched >= total:
                return
            if len(page_items) < page_size:
                return
            start += page_size

    async def search(self, itemtype: str, **kwargs: Any) -> dict:
        return await self._request("GET", f"search/{_hl_route(itemtype)}", params=_boolify_params(kwargs))

    async def create_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> Any:
        if isinstance(input_data, dict):
            body = dict(input_data)
            body.update(kwargs)
        else:
            body = input_data
        return await self._request("POST", _hl_route(itemtype), json=body)

    async def update_item(self, itemtype: str, input_data: Any, **kwargs: Any) -> list:
        if isinstance(input_data, dict) and "id" in input_data:
            item_id = input_data["id"]
            body = {k: v for k, v in input_data.items() if k != "id"}
            body.update(kwargs)
            return await self._request("PATCH", f"{_hl_route(itemtype)}/{item_id}", json=body)
        payload: dict = {"input": input_data}
        payload.update(kwargs)
        return await self._request("PATCH", _hl_route(itemtype), json=payload)

    async def delete_item(self, itemtype: str, input_data: Any, force_purge: bool = False, history: bool = True) -> list:
        if isinstance(input_data, dict) and "id" in input_data:
            item_id = input_data["id"]
            return await self._request("DELETE", f"{_hl_route(itemtype)}/{item_id}")
        return await self._request("DELETE", _hl_route(itemtype), json={"input": input_data})

    async def get_sub_items(self, itemtype: str, item_id: int, sub_itemtype: str, **kwargs: Any) -> list:
        return await self._request("GET", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", params=_boolify_params(kwargs))

    async def add_sub_item(self, itemtype: str, item_id: int, sub_itemtype: str, input_data: dict, **kwargs: Any) -> dict:
        body = dict(input_data)
        body.update(kwargs)
        return await self._request("POST", f"{_hl_route(itemtype)}/{item_id}/{_hl_route(sub_itemtype)}", json=body)

    @property
    def version(self) -> Optional[GLPIVersion]:
        return self._version

close `async` ¶

close() -> None

Close the underlying aiohttp session.

Source code in glpi_utils/oauth.py

async def close(self) -> None:
    """Close the underlying aiohttp session."""
    if self._http and not self._http.closed:
        await self._http.close()
    self._token.clear()

authenticate `async` ¶

authenticate(username: Optional[str] = None, password: Optional[str] = None) -> None

Obtain an OAuth2 Bearer token asynchronously.

Source code in glpi_utils/oauth.py

async def authenticate(
    self,
    username: Optional[str] = None,
    password: Optional[str] = None,
) -> None:
    """Obtain an OAuth2 Bearer token asynchronously."""
    import aiohttp

    if not self._client_id:
        raise GlpiAuthError("client_id is required. Set GLPI_OAUTH_CLIENT_ID.")

    username = username or self._username
    password = password or self._password

    if username and password:
        payload = {
            "grant_type": "password",
            "client_id": self._client_id,
            "client_secret": self._client_secret or "",
            "username": username,
            "password": password,
            "scope": "api",
        }
        grant = "password"
    else:
        if not self._client_secret:
            raise GlpiAuthError(
                "client_secret is required for client_credentials grant."
            )
        payload = {
            "grant_type": "client_credentials",
            "client_id": self._client_id,
            "client_secret": self._client_secret,
            "scope": "api",
        }
        grant = "client_credentials"

    log.debug("Async OAuth2 token request (grant=%s)", grant)

    http = self._get_http()
    timeout = aiohttp.ClientTimeout(total=self._timeout)

    try:
        async with http.post(
            self._token_url, data=payload, timeout=timeout
        ) as resp:
            body = await resp.json(content_type=None)
            if resp.status not in (200, 201) or "access_token" not in body:
                error = body.get("error", str(resp.status))
                desc  = body.get("error_description", "")
                raise GlpiAuthError(
                    f"OAuth2 token request failed ({error}): {desc}",
                    status_code=resp.status,
                    error_code=error,
                )
            self._token.store(body["access_token"], int(body.get("expires_in", 3600)))
            log.debug("Async OAuth2 token obtained.")
    except aiohttp.ClientConnectorError as exc:
        raise GlpiConnectionError(f"Cannot reach GLPI at {self._url}: {exc}") from exc

GlpiOAuthClient

glpi_utils.oauth.GlpiOAuthClient ¶

version property ¶

close ¶

item ¶

authenticate ¶

get_version ¶

get_item ¶

get_all_items ¶

get_all_pages ¶

iter_pages ¶

search ¶

create_item ¶

update_item ¶

delete_item ¶

get_sub_items ¶

add_sub_item ¶

glpi_utils.oauth.AsyncGlpiOAuthClient ¶

close async ¶

authenticate async ¶

version `property` ¶

close `async` ¶

authenticate `async` ¶