feat: add FastAPI anonymizer server

presidio-anonymizer/README.md

@@ -202,3 +202,14 @@ docker-compose up -d
 
 Follow the [API Spec](https://microsoft.github.io/presidio/api-docs/api-docs.html#tag/Anonymizer) for the
 Anonymizer REST API reference details
+
+#### Optional FastAPI server
+
+The anonymizer source tree also includes a FastAPI server with the same REST
+endpoints as the default Flask server. From the `presidio-anonymizer` source
+directory, install the optional dependencies and run it with Uvicorn:
+
+```sh
+pip install "presidio-anonymizer[fastapi]"
+uvicorn fastapi_app:app --host 0.0.0.0 --port 3000
+```

presidio-anonymizer/fastapi_app.py

@@ -0,0 +1,148 @@
+"""FastAPI REST API server for anonymizer."""
+
+import json
+import logging
+import os
+from logging.config import fileConfig
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import JSONResponse, Response
+from presidio_anonymizer import AnonymizerEngine, DeanonymizeEngine
+from presidio_anonymizer.entities import InvalidParamError
+from presidio_anonymizer.services.app_entities_convertor import AppEntitiesConvertor
+
+DEFAULT_PORT = "3000"
+
+LOGGING_CONF_FILE = "logging.ini"
+
+WELCOME_MESSAGE = r"""
+ _______  _______  _______  _______ _________ ______  _________ _______
+(  ____ )(  ____ )(  ____ \(  ____ \\__   __/(  __  \ \__   __/(  ___  )
+| (    )|| (    )|| (    \/| (    \/   ) (   | (  \  )   ) (   | (   ) |
+| (____)|| (____)|| (__    | (_____    | |   | |   ) |   | |   | |   | |
+|  _____)|     __)|  __)   (_____  )   | |   | |   | |   | |   | |   | |
+| (      | (\ (   | (            ) |   | |   | |   ) |   | |   | |   | |
+| )      | ) \ \__| (____/\/\____) |___) (___| (__/  )___) (___| (___) |
+|/       |/   \__/(_______/\_______)\_______/(______/ \_______/(_______)
+"""
+
+
+class Server:
+    """FastAPI server for anonymizer."""
+
+    def __init__(self) -> None:
+        fileConfig(Path(Path(__file__).parent, LOGGING_CONF_FILE))
+        self.logger = logging.getLogger("presidio-anonymizer")
+        self.logger.setLevel(os.environ.get("LOG_LEVEL", self.logger.level))
+        self.app = FastAPI(title="Presidio Anonymizer")
+        self.logger.info("Starting anonymizer engine")
+        self.anonymizer = AnonymizerEngine()
+        self.deanonymize = DeanonymizeEngine()
+        self.logger.info(WELCOME_MESSAGE)
+        self._add_routes()
+        self._add_error_handlers()
+
+    def _add_routes(self) -> None:
+        @self.app.get("/health", response_class=Response)
+        def health() -> str:
+            """Return basic health probe result."""
+            return "Presidio Anonymizer service is up"
+
+        @self.app.post("/anonymize")
+        async def anonymize(request: Request) -> Response:
+            content = await self._json_request_body(request)
+
+            anonymizers_config = AppEntitiesConvertor.operators_config_from_json(
+                content.get("anonymizers")
+            )
+            if AppEntitiesConvertor.check_custom_operator(anonymizers_config):
+                raise HTTPException(
+                    status_code=400, detail="Custom type anonymizer is not supported"
+                )
+
+            analyzer_results = AppEntitiesConvertor.analyzer_results_from_json(
+                content.get("analyzer_results")
+            )
+            anonymizer_result = self.anonymizer.anonymize(
+                text=content.get("text", ""),
+                analyzer_results=analyzer_results,
+                operators=anonymizers_config,
+            )
+            return Response(
+                content=anonymizer_result.to_json(), media_type="application/json"
+            )
+
+        @self.app.post("/deanonymize")
+        async def deanonymize(request: Request) -> Response:
+            content = await self._json_request_body(request)
+
+            deanonymize_entities = AppEntitiesConvertor.deanonymize_entities_from_json(
+                content
+            )
+            deanonymize_config = AppEntitiesConvertor.operators_config_from_json(
+                content.get("deanonymizers")
+            )
+            deanonymized_response = self.deanonymize.deanonymize(
+                text=content.get("text", ""),
+                entities=deanonymize_entities,
+                operators=deanonymize_config,
+            )
+            return Response(
+                content=deanonymized_response.to_json(),
+                media_type="application/json",
+            )
+
+        @self.app.get("/anonymizers")
+        def anonymizers() -> list[str]:
+            """Return a list of supported anonymizers."""
+            return self.anonymizer.get_anonymizers()
+
+        @self.app.get("/deanonymizers")
+        def deanonymizers() -> list[str]:
+            """Return a list of supported deanonymizers."""
+            return self.deanonymize.get_deanonymizers()
+
+    async def _json_request_body(self, request: Request) -> dict[str, Any]:
+        try:
+            content = await request.json()
+        except json.JSONDecodeError:
+            raise HTTPException(status_code=400, detail="Invalid request json")
+
+        if not content or not isinstance(content, dict):
+            raise HTTPException(status_code=400, detail="Invalid request json")
+        return content
+
+    def _add_error_handlers(self) -> None:
+        @self.app.exception_handler(InvalidParamError)
+        def invalid_param(_: Request, err: InvalidParamError) -> JSONResponse:
+            self.logger.warning(
+                "Request failed with parameter validation error: %s", err.err_msg
+            )
+            return JSONResponse({"error": err.err_msg}, status_code=422)
+
+        @self.app.exception_handler(HTTPException)
+        def http_exception(_: Request, err: HTTPException) -> JSONResponse:
+            return JSONResponse({"error": err.detail}, status_code=err.status_code)
+
+        @self.app.exception_handler(Exception)
+        def server_error(_: Request, err: Exception) -> JSONResponse:
+            self.logger.error("A fatal error occurred during execution: %s", err)
+            return JSONResponse({"error": "Internal server error"}, status_code=500)
+
+
+def create_app() -> FastAPI:
+    """Create the FastAPI application."""
+    server = Server()
+    return server.app
+
+
+app = create_app()
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    port = int(os.environ.get("PORT", DEFAULT_PORT))
+    uvicorn.run(app, host="0.0.0.0", port=port)

presidio-anonymizer/pyproject.toml

@@ -31,6 +31,10 @@ server = [
     "gunicorn (>=20.0.0,<26.0.0); platform_system != 'Windows'",
     "waitress (>=2.0.0,<4.0.0); platform_system == 'Windows'"
 ]
+fastapi = [
+    "fastapi (>=0.115.0,<1.0.0)",
+    "uvicorn (>=0.32.0,<1.0.0)"
+]
 ahds = [
     "azure-identity (>=1.25.3,<2.0.0)",
     "azure-health-deidentification (>=1.1.0b1,<2.0.0)"
@@ -45,6 +49,8 @@ pytest-mock = "*"
 python-dotenv = "*"
 pre_commit = "*"
 diff-cover = "*"
+fastapi = "*"
+httpx = "*"
 
 [tool.coverage.run]
 relative_files = true

presidio-anonymizer/tests/test_fastapi_app.py

@@ -0,0 +1,120 @@
+"""Tests for the FastAPI anonymizer server."""
+
+from importlib import util
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("httpx")
+fastapi_testclient = pytest.importorskip("fastapi.testclient")
+TestClient = fastapi_testclient.TestClient
+
+
+def _load_fastapi_app():
+    module_path = Path(__file__).parents[1] / "fastapi_app.py"
+    spec = util.spec_from_file_location("presidio_anonymizer_fastapi_app", module_path)
+    module = util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.app
+
+
+def test_health_endpoint_returns_service_status():
+    """Health endpoint mirrors the existing service status response."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.get("/health")
+
+    assert response.status_code == 200
+    assert response.text == "Presidio Anonymizer service is up"
+
+
+def test_anonymize_endpoint_returns_engine_response():
+    """Anonymize endpoint returns the anonymizer engine JSON response."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.post(
+        "/anonymize",
+        json={
+            "text": "My name is Jane",
+            "analyzer_results": [
+                {
+                    "start": 11,
+                    "end": 15,
+                    "score": 0.8,
+                    "entity_type": "PERSON",
+                }
+            ],
+            "anonymizers": {
+                "DEFAULT": {"type": "replace", "new_value": "<ANONYMIZED>"}
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["text"] == "My name is <ANONYMIZED>"
+
+
+def test_empty_json_body_returns_flask_compatible_error_shape():
+    """Empty JSON requests keep the existing error response shape."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.post("/anonymize", json={})
+
+    assert response.status_code == 400
+    assert response.json() == {"error": "Invalid request json"}
+
+
+@pytest.mark.parametrize("request_kwargs", [{}, {"content": "not json"}])
+def test_invalid_json_body_returns_flask_compatible_error_shape(request_kwargs):
+    """Missing and invalid JSON requests keep the existing error response shape."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.post("/anonymize", **request_kwargs)
+
+    assert response.status_code == 400
+    assert response.json() == {"error": "Invalid request json"}
+
+
+def test_anonymizers_endpoint_returns_supported_operators():
+    """Anonymizers endpoint exposes built-in anonymizer operators."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.get("/anonymizers")
+
+    assert response.status_code == 200
+    assert "replace" in response.json()
+
+
+def test_deanonymize_endpoint_returns_engine_response():
+    """Deanonymize endpoint returns the deanonymizer engine JSON response."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.post(
+        "/deanonymize",
+        json={
+            "text": "My name is Jane",
+            "anonymizer_results": [
+                {
+                    "start": 11,
+                    "end": 15,
+                    "entity_type": "PERSON",
+                    "text": "Jane",
+                    "operator": "keep",
+                }
+            ],
+            "deanonymizers": {"DEFAULT": {"type": "deanonymize_keep"}},
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["text"] == "My name is Jane"
+
+
+def test_deanonymizers_endpoint_returns_supported_operators():
+    """Deanonymizers endpoint exposes built-in deanonymizer operators."""
+    client = TestClient(_load_fastapi_app())
+
+    response = client.get("/deanonymizers")
+
+    assert response.status_code == 200
+    assert "deanonymize_keep" in response.json()