Support both Digest and API key authentication

README.md

@@ -1,3 +1,5 @@
 # PyPrusaLink
 
 Python library to interact with PrusaLink API v2.
+
+This fork adds support for Digest authentication - the method Prusa recommends for their resin printers (SL1, SL1S).

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pyprusalink"
-version = "1.1.0"
+version = "1.2.0"
 license     = {text = "Apache-2.0"}
 description = "Library to interact with PrusaLink v2"
 readme = "README.md"
@@ -20,7 +20,7 @@ classifiers = [
   "Topic :: Home Automation",
 ]
 dependencies = [
-  "aiohttp",
+  "httpx",
 ]
 
 [tool.setuptools]

pyprusalink/__init__.py

@@ -3,9 +3,27 @@
 
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import TypedDict
 
-from aiohttp import ClientResponse, ClientSession
+from httpx import AsyncClient, DigestAuth, Response
+
+from .const import (
+    API_KEY,
+    API_KEY_AUTH,
+    AUTH,
+    AUTH_TYPE,
+    DIGEST_AUTH,
+    HOST,
+    PASSWORD,
+    USER,
+)
+from .types import (
+    FileInfo,
+    FilesInfo,
+    JobInfo,
+    LinkConfiguration,
+    PrinterInfo,
+    VersionInfo,
+)
 
 
 class PrusaLinkError(Exception):
@@ -20,57 +38,19 @@ class Conflict(PrusaLinkError):
     """Error to indicate the command hit a conflict."""
 
 
-class VersionInfo(TypedDict):
-    """Version data."""
-
-    api: str
-    server: str
-    text: str
-    hostname: str
-
-
-class PrinterInfo(TypedDict):
-    """Printer data."""
-
-    telemetry: dict
-    temperature: dict
-    state: dict
-
-
-class JobInfo(TypedDict):
-    """Job data."""
-
-    state: str
-    job: dict | None
-
-
-class FileInfo(TypedDict):
-    """File data."""
-
-    name: str
-    origin: str
-    size: int
-    refs: dict
-
-
-class FilesInfo(TypedDict):
-    """Files data."""
-
-    files: list[FileInfo]
-
-
 class PrusaLink:
     """Wrapper for the Prusalink API.
 
     Data format can be found here:
     https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/master/lib/WUI/link_content/basic_gets.cpp
     """
 
-    def __init__(self, session: ClientSession, host: str, api_key: str) -> None:
+    def __init__(self, client: AsyncClient, config: LinkConfiguration) -> None:
         """Initialize the PrusaLink class."""
-        self._session = session
-        self.host = host
-        self._api_key = api_key
+
+        self.host = config[HOST]
+        self.auth = config[AUTH]
+        self.client = client
 
     async def cancel_job(self) -> None:
         """Cancel the current job."""
@@ -94,53 +74,65 @@ async def pause_job(self) -> None:
     async def get_version(self) -> VersionInfo:
         """Get the version."""
         async with self.request("GET", "api/version") as response:
-            return await response.json()
+            return response.json()
 
     async def get_printer(self) -> PrinterInfo:
         """Get the printer."""
         async with self.request("GET", "api/printer") as response:
-            return await response.json()
+            return response.json()
 
     async def get_job(self) -> JobInfo:
         """Get current job."""
         async with self.request("GET", "api/job") as response:
-            return await response.json()
+            return response.json()
 
     async def get_file(self, path: str) -> FileInfo:
         """Get specific file info."""
         async with self.request("GET", f"api/files{path}") as response:
-            return await response.json()
+            return response.json()
 
     async def get_files(self) -> FilesInfo:
         """Get all files."""
         async with self.request("GET", "api/files?recursive=true") as response:
-            return await response.json()
+            return response.json()
 
     async def get_small_thumbnail(self, path: str) -> bytes:
         """Get a small thumbnail."""
         async with self.request("GET", f"thumb/s{path}") as response:
-            return await response.read()
+            return response.read()
 
     async def get_large_thumbnail(self, path: str) -> bytes:
         """Get a large thumbnail."""
         async with self.request("GET", f"thumb/l{path}") as response:
-            return await response.read()
+            return response.read()
 
     @asynccontextmanager
     async def request(
         self, method: str, path: str, json: dict | None = None
-    ) -> AsyncGenerator[ClientResponse, None]:
+    ) -> AsyncGenerator[Response, None]:
         """Make a request to the PrusaLink API."""
         url = f"{self.host}/{path}"
-        headers = {"X-Api-Key": self._api_key}
-
-        async with self._session.request(
-            method, url, headers=headers, json=json
-        ) as response:
-            if response.status == 401:
+        client = self.client
+
+        async with client:
+            if self.auth[AUTH_TYPE] == DIGEST_AUTH:
+                auth = DigestAuth(self.auth[USER], self.auth[PASSWORD])
+                response = await client.request(method, url, json=json, auth=auth)
+
+            elif self.auth[AUTH_TYPE] == API_KEY_AUTH:
+                headers = {"X-Api-Key": self.auth[API_KEY]}
+                response = await client.request(
+                    method,
+                    url,
+                    json=json,
+                    headers=headers,
+                )
+
+            if response.status_code == 401:
                 raise InvalidAuth()
-            if response.status == 409:
+            elif response.status_code == 409:
                 raise Conflict()
 
             response.raise_for_status()
+
             yield response

pyprusalink/const.py

@@ -0,0 +1,8 @@
+API_KEY = "apiKey"
+API_KEY_AUTH = "ApiKeyAuth"
+AUTH = "auth"
+AUTH_TYPE = "authType"
+DIGEST_AUTH = "DigestAuth"
+HOST = "host"
+PASSWORD = "password"
+USER = "user"

pyprusalink/types.py

@@ -0,0 +1,64 @@
+from typing import TypedDict
+
+from .const import API_KEY_AUTH, DIGEST_AUTH
+
+
+class VersionInfo(TypedDict):
+    """Version data."""
+
+    api: str
+    server: str
+    text: str
+    hostname: str
+
+
+class PrinterInfo(TypedDict):
+    """Printer data."""
+
+    telemetry: dict
+    temperature: dict
+    state: dict
+
+
+class JobInfo(TypedDict):
+    """Job data."""
+
+    state: str
+    job: dict | None
+
+
+class FileInfo(TypedDict):
+    """File data."""
+
+    name: str
+    origin: str
+    size: int
+    refs: dict
+
+
+class FilesInfo(TypedDict):
+    """Files data."""
+
+    files: list[FileInfo]
+
+
+class UserAuth(TypedDict):
+    """Authentication via `user` + `password` digest"""
+
+    type: DIGEST_AUTH
+    user: str
+    password: str
+
+
+class ApiKeyAuth(TypedDict):
+    """Authentication via api key"""
+
+    type: API_KEY_AUTH
+    apiKey: str
+
+
+class LinkConfiguration(TypedDict):
+    """Configuration shape for PrusaLink"""
+
+    host: str
+    auth: UserAuth | ApiKeyAuth

shell.nix

@@ -0,0 +1,13 @@
+{ pkgs ? import <nixpkgs> {} }:
+
+pkgs.mkShell {
+  packages = [
+    (pkgs.python3.withPackages (ps: [
+      ps.httpx
+      ps.isort
+      ps.flake8
+      ps.black
+    ]))
+  ];
+}
+