add: python-bsonjs response class (with C ffi), tests

fastapi/responses.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+import json
+
 from starlette.responses import FileResponse as FileResponse  # noqa
 from starlette.responses import HTMLResponse as HTMLResponse  # noqa
 from starlette.responses import JSONResponse as JSONResponse  # noqa
@@ -26,6 +28,12 @@ except ImportError:  # pragma: nocover
     bson = None
 
 
+try:
+    import bsonjs  # type: ignore[import-untyped]
+except ImportError:  # pragma: nocover
+    bsonjs = None
+
+
 class UJSONResponse(JSONResponse):
     """
     JSON response using the high-performance ujson library to serialize data to JSON.
@@ -69,3 +77,18 @@ class BSONResponse(Response):
     def render(self, content: Any) -> Any:
         assert bson is not None, "bson must be installed to use BSONResponse"
         return bson.dumps(content)
+
+
+class BSONJSResponse(BSONResponse):
+    """
+    BSON response using the C-backed python-bsonjs library to serialize BSON
+
+    Read more about it in the
+    [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/).
+    """
+
+    media_type = "application/bson"
+
+    def render(self, content: Any) -> Any:
+        assert bsonjs is not None, "bsonjs must be installed to use BSONJSResponse"
+        return bsonjs.loads(json.dumps(content))

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
     "starlette>=0.40.0,<0.47.0",
     "pydantic>=1.7.4,!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0",
     "typing-extensions>=4.8.0",
+    "python-bsonjs>=0.5.0",
 ]
 
 [project.urls]
@@ -94,6 +95,8 @@ all = [
     "orjson >=3.2.1",
     # For BSONResponse
     "bson >=0.5.10",
+    # For BSONJSResponse
+    "python-bsonjs >=0.6.0",
     # To validate email fields
     "email-validator >=2.0.0",
     # Uvicorn with uvloop

tests/test_bsonjs_response_class.py

@@ -0,0 +1,23 @@
+import bsonjs
+import json
+
+from fastapi import FastAPI
+from fastapi.responses import BSONJSResponse
+from fastapi.testclient import TestClient
+
+app = FastAPI(default_response_class=BSONJSResponse)
+
+
+@app.get("/bsonjs_keys")
+def get_bsonjs_serialized_data():
+    return {"key": "Hello World", 1: 1}
+
+
+client = TestClient(app)
+
+
+def test_bsonjs_serialized_data():
+    with client:
+        response = client.get("/bsonjs_keys")
+
+    assert response.content == bsonjs.loads(json.dumps({"key": "Hello World", 1: 1}))