Add FastAPI module

Author: Tranquility2

modules/fastapi/README.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.fastapi.FastAPIContainer
+.. title:: testcontainers.fastapi.FastAPIContainer

modules/fastapi/testcontainers/fastapi/__init__.py

@@ -0,0 +1,43 @@
+from typing import Optional
+
+import httpx
+
+from testcontainers.core.generic import SrvContainer
+
+
+class FastAPIContainer(SrvContainer):
+    """
+    FastAPI container that is based on a custom image.
+
+    Example:
+
+        .. doctest::
+
+            >>> from testcontainers.fastapi import FastAPIContainer
+            >>> from testcontainers.core.waiting_utils import wait_for_logs
+
+            >>> with FastAPIContainer(path="./modules/fastapi/tests/sample", port=80, tag="fastapi:latest") as fastapi:
+            ...     delay = wait_for_logs(fastapi, "Uvicorn running on http://0.0.0.0:80")
+            ...     client = fastapi.get_client()
+            ...     response = client.get("/")
+            ...     assert response.status_code == 200
+            ...     assert response.json() == {"Status": "Working"}
+    """
+
+    def __init__(self, path: str, port: int, tag: Optional[str] = None, image_cleanup: bool = True) -> None:
+        """
+        :param path: Path to the FastAPI application.
+        :param port: Port to expose the FastAPI application.
+        :param tag: Tag for the image to be built (default: None).
+        :param image_cleanup: Clean up the image after the container is stopped (default: True).
+        """
+        super().__init__(path, port, tag, image_cleanup)
+
+    def get_api_url(self) -> str:
+        return self._create_connection_url() + "/api/v1/"
+
+    def get_client(self) -> httpx.Client:
+        return httpx.Client(base_url=self.get_api_url())
+
+    def get_stdout(self) -> str:
+        return self.get_logs()[0].decode("utf-8")

modules/fastapi/tests/sample/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9
+
+WORKDIR /app
+
+RUN pip install fastapi
+
+COPY ./app /app
+
+EXPOSE 80
+
+CMD ["fastapi", "run", "main.py", "--port", "80"]

modules/fastapi/tests/sample/app/__init__.py

@@ -0,0 +1,8 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+
+@app.get("/api/v1/")
+def read_root():
+    return {"Status": "Working"}

modules/fastapi/tests/sample/app/main.py

@@ -0,0 +1,33 @@
+import re
+import pytest
+
+from testcontainers.fastapi import FastAPIContainer
+
+
+def test_fastapi_container():
+    with FastAPIContainer(
+        path="./modules/fastapi/tests/sample", port=80, tag="fastapi:test", image_cleanup=False
+    ) as fastapi:
+        assert fastapi.get_container_host_ip() == "localhost"
+        assert fastapi.internal_port == 80
+        assert re.match(r"http://localhost:\d+/api/v1/", fastapi.get_api_url())
+        assert fastapi.get_client().get("/").status_code == 200
+        assert fastapi.get_client().get("/").json() == {"Status": "Working"}
+
+
+def test_fastapi_container_no_tag():
+    with FastAPIContainer(path="./modules/fastapi/tests/sample", port=80, image_cleanup=False) as fastapi:
+        assert fastapi.get_client().get("/").status_code == 200
+        assert fastapi.get_client().get("/").json() == {"Status": "Working"}
+
+
+def test_fastapi_container_no_port():
+    with pytest.raises(TypeError):
+        with FastAPIContainer(path="./modules/fastapi/tests/sample", tag="fastapi:test", image_cleanup=False):
+            pass
+
+
+def test_fastapi_container_no_path():
+    with pytest.raises(TypeError):
+        with FastAPIContainer(port=80, tag="fastapi:test", image_cleanup=True):
+            pass

modules/fastapi/tests/test_fastapi.py

@@ -57,7 +57,8 @@ packages = [
     { include = "testcontainers", from = "modules/registry" },
     { include = "testcontainers", from = "modules/selenium" },
     { include = "testcontainers", from = "modules/vault" },
-    { include = "testcontainers", from = "modules/weaviate" }
+    { include = "testcontainers", from = "modules/weaviate" },
+    { include = "testcontainers", from = "modules/fastapi" },
 ]
 
 [tool.poetry.urls]