fix: add support for application/json-seq & application/x-ndjson

README.md

@@ -1,6 +1,6 @@
 # snap-http
 
-![check](https://github.com/perfect5th/snap-http/actions/workflows/test.yml/badge.svg)
+![check](https://github.com/canonical/snap-http/actions/workflows/test.yml/badge.svg)
 
 snap-http is a Python library used to interact with snapd's REST API, allowing you to
 programmatically install and manage snaps in your Python applications. It has no dependencies
@@ -14,7 +14,7 @@ pip install snap-http
 
 ## Usage
 
-Take a look at the [api](https://github.com/Perfect5th/snap-http/blob/main/snap_http/api.py) module
+Take a look at the [api](https://github.com/canonical/snap-http/blob/main/snap_http/api/__init__.py) module
 to see what methods are available. Here's a couple examples:
 
 ### List installed snaps

pyproject.toml

@@ -2,7 +2,7 @@
 name = "snap-http"
 version = "1.10.1"
 description = "A library for interacting with snapd via its REST API."
-authors = ["Mitch Burton <mitch.burton@canonical.com>"]
+authors = ["Mitch Burton <mitch.burton@canonical.com>", "Stephen Mwangi <mail@stephenmwangi.com>"]
 readme = "README.md"
 exclude = [
     "tests/"

snap_http/http.py

@@ -98,15 +98,25 @@ def _make_request(
     if response.status >= 400:
         raise SnapdHttpException(response_body)
 
-    response_type = response.getheader("Content-Type")
-    if response_type == "application/json":
+    content_type = response.getheader("Content-Type")
+    if content_type == "application/json":
         return json.loads(response_body)
+    elif content_type in ("application/json-seq", "application/x-ndjson"):
+        records = [
+            json.loads(record)
+            for record in response_body.split(b"\x1e")
+            if record.strip()
+        ]
+        return _build_envelope(response, records)
     else:  # other types like application/x.ubuntu.assertion
-        response_code = response.getcode()
-        is_async = response_code == 202
-        return {
-            "type": "async" if is_async else "sync",
-            "status_code": response_code,
-            "status": responses[response_code],
-            "result": response_body,
-        }
+        return _build_envelope(response, response_body)
+
+
+def _build_envelope(response: HTTPResponse, result: Any) -> Dict[str, Any]:
+    response_code = response.getcode()
+    return {
+        "type": "async" if response_code == 202 else "sync",
+        "status_code": response_code,
+        "status": responses[response_code],
+        "result": result,
+    }

tests/integration/test_snaps.py

@@ -169,3 +169,19 @@ def test_sideload_multiple_snaps(
     assert is_snap_installed("hello-world") is True
 
     wait_for(snap_http.remove_all)(["test-snap", "hello-world"])
+
+
+def test_logs(test_snap):
+    """Test getting snap logs."""
+    wait_for(snap_http.start)("test-snap.bye-svc")
+
+    response = snap_http.logs(names=["test-snap"], entries=5)
+    assert response.status_code == 200
+
+    assert len(response.result) > 0
+    for entry in response.result:
+        assert "timestamp" in entry
+        assert "message" in entry
+        assert "sid" in entry
+        assert "pid" in entry
+    assert any("test-snap.bye-svc" in entry["message"] for entry in response.result)

tests/unit/api/test_snaps.py

@@ -941,7 +941,10 @@ def test_logs(monkeypatch, names, entries, expected_params):
         type="sync",
         status_code=200,
         status="OK",
-        result=[{"title": "placeholder1"}, {"title": "placeholder2"}],
+        result=[
+            {"timestamp": "2026-03-25T04:57:10Z", "message": "hello", "sid": "systemd", "pid": "1"},
+            {"timestamp": "2026-03-25T04:57:11Z", "message": "world", "sid": "systemd", "pid": "2"},
+        ],
     )
 
     def mock_get(path, query_params):

tests/unit/test_http.py

@@ -149,6 +149,29 @@ def test_get_non_json_data(use_snapd_response, monkeypatch):
     )
 
 
+@pytest.mark.parametrize("content_type", ["application/x-ndjson", "application/json-seq"])
+def test_get_ndjson_data(use_snapd_response, monkeypatch, content_type):
+    """`http.get` parses ndjson responses into a list of dicts."""
+    monkeypatch.setattr(http, "SNAPD_SOCKET", FAKE_SNAPD_SOCKET)
+    mock_response = (
+        b'\x1e{"timestamp":"2026-03-25T04:57:10Z","message":"hello","sid":"systemd","pid":"1"}\n'
+        b'\x1e{"timestamp":"2026-03-25T04:57:11Z","message":"world","sid":"systemd","pid":"2"}\n'
+    )
+    use_snapd_response(200, mock_response, content_type)
+
+    result = http.get("/logs")
+
+    assert result == types.SnapdResponse(
+        type="sync",
+        status_code=200,
+        status="OK",
+        result=[
+            {"timestamp": "2026-03-25T04:57:10Z", "message": "hello", "sid": "systemd", "pid": "1"},
+            {"timestamp": "2026-03-25T04:57:11Z", "message": "world", "sid": "systemd", "pid": "2"},
+        ],
+    )
+
+
 def test_get_returns_a_warning(use_snapd_response, monkeypatch):
     """`http.get` returns a `types.SnapdResponse.`"""
     monkeypatch.setattr(http, "SNAPD_SOCKET", FAKE_SNAPD_SOCKET)