feat(ext): fastapi network context

pyproject.toml

@@ -125,6 +125,8 @@ test = [  # `test` GitHub Action jobs use this
     "ape-vyper>=0.8.10,<0.9",  # Needed for compiling test contracts
     "vyper>=0.4.3,<0.5",  # Avoid having to download Vyper binaries
     "ape-solidity>=0.8.5,<0.9",  # Needed for compiling test contracts
+    "fastapi",  # Needed for ext/fastapi tests
+    "httpx",  # Needed for FastAPI TestClient
 ]
 lint = [  # `lint` GitHub Action jobs use this
     "ruff>=0.12.0",  # Unified linter and formatter

src/ape/ext/__init__.py

@@ -0,0 +1 @@
+# ape.ext — External framework integrations for Ape

src/ape/ext/fastapi.py

@@ -0,0 +1,110 @@
+import functools
+import inspect
+from collections.abc import Callable
+from typing import Any
+
+
+def network_context() -> Callable:
+    """
+    A decorator that wraps a FastAPI route handler to automatically
+    enter the correct Ape network context.
+
+    ``ecosystem`` and ``network`` are injected as FastAPI Query parameters
+    automatically — you do not need to declare them in the function signature.
+    If you do declare them, they will still be passed through to your function.
+
+    Example (without declaring ecosystem/network)::
+
+        @app.post("/faucet/{token}")
+        @network_context()
+        def send_token(token: str, receiver: str, amount: int) -> str:
+            # Ape is connected to the network requested via query string
+            # e.g. ?ecosystem=ethereum&network=mainnet
+            ...
+
+    Example (with ecosystem/network in signature)::
+
+        @app.post("/faucet/{token}")
+        @network_context()
+        def send_token(token: str, ecosystem: str, network: str) -> str:
+            print(f"Connected to {ecosystem}:{network}")
+            ...
+
+    Raises:
+        :class:`~ape.exceptions.NetworkError`: When the given ecosystem/network
+            combination does not match any known network.
+        ``ImportError``: When ``fastapi`` is not installed.
+    """
+
+    def decorator(func: Callable) -> Callable:
+        original_params = inspect.signature(func).parameters
+
+        @functools.wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            ecosystem = kwargs.get("ecosystem")
+            network = kwargs.get("network")
+
+            network_choice = f"{ecosystem}:{network}"
+
+            # Import here so fastapi remains an optional dependency.
+            # ape.networks is also imported here to avoid circular imports
+            # at module load time.
+            from ape import networks
+
+            with networks.parse_network_choice(network_choice):
+                # Only strip ecosystem/network from kwargs if the original
+                # function does not declare them — otherwise pass them through.
+                call_kwargs = dict(kwargs)
+                if "ecosystem" not in original_params:
+                    call_kwargs.pop("ecosystem", None)
+                if "network" not in original_params:
+                    call_kwargs.pop("network", None)
+
+                return func(*args, **call_kwargs)
+
+        try:
+            from fastapi import Query
+        except ImportError as e:
+            raise ImportError(
+                "fastapi is required to use ape.ext.fastapi. Install it with: pip install fastapi"
+            ) from e
+
+        original_sig = inspect.signature(func)
+        extra_params: list[inspect.Parameter] = []
+
+        if "ecosystem" not in original_params:
+            extra_params.append(
+                inspect.Parameter(
+                    "ecosystem",
+                    kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    default=Query(..., description="Ape ecosystem name, e.g. 'ethereum'"),
+                    annotation=str,
+                )
+            )
+
+        if "network" not in original_params:
+            extra_params.append(
+                inspect.Parameter(
+                    "network",
+                    kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    default=Query(..., description="Ape network name, e.g. 'mainnet'"),
+                    annotation=str,
+                )
+            )
+
+        if extra_params:
+            existing = list(original_sig.parameters.values())
+            # Insert extra params before **kwargs (VAR_KEYWORD) if present,
+            # since **kwargs must always be the last parameter in a signature.
+            insert_at = len(existing)
+            for i, p in enumerate(existing):
+                if p.kind == inspect.Parameter.VAR_KEYWORD:
+                    insert_at = i
+                    break
+
+            new_params = existing[:insert_at] + extra_params + existing[insert_at:]
+            wrapper.__signature__ = original_sig.replace(parameters=new_params)  # type: ignore[attr-defined]
+
+        return wrapper
+
+    return decorator

tests/functional/test_ext_fastapi.py

@@ -0,0 +1,174 @@
+import inspect
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Skip the entire module if fastapi is not installed.
+fastapi = pytest.importorskip("fastapi")
+
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+from ape.ext.fastapi import network_context  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_app_and_client(handler):
+    app = FastAPI()
+    app.get("/test")(handler)
+    return TestClient(app, raise_server_exceptions=True)
+
+
+def _mock_network_ctx():
+    ctx = MagicMock()
+    ctx.__enter__ = MagicMock(return_value=None)
+    ctx.__exit__ = MagicMock(return_value=False)
+    return ctx
+
+
+# ---------------------------------------------------------------------------
+# Test 1 — decorator adds ecosystem/network to signature automatically
+# ---------------------------------------------------------------------------
+
+
+class TestSignaturePatch:
+    def test_ecosystem_and_network_added_when_not_in_original(self):
+
+        @network_context()
+        def handler(token: str) -> dict:
+            return {"token": token}
+
+        sig = inspect.signature(handler)
+        assert "ecosystem" in sig.parameters
+        assert "network" in sig.parameters
+
+    def test_ecosystem_and_network_not_duplicated_when_already_present(self):
+
+        @network_context()
+        def handler(token: str, ecosystem: str, network: str) -> dict:
+            return {}
+
+        sig = inspect.signature(handler)
+        param_names = list(sig.parameters.keys())
+        assert param_names.count("ecosystem") == 1
+        assert param_names.count("network") == 1
+
+    def test_original_params_preserved_in_signature(self):
+
+        @network_context()
+        def handler(token: str, amount: int) -> dict:
+            return {}
+
+        sig = inspect.signature(handler)
+        assert "token" in sig.parameters
+        assert "amount" in sig.parameters
+
+
+# ---------------------------------------------------------------------------
+# Test 2 — network context is entered with the correct choice string
+# ---------------------------------------------------------------------------
+
+
+class TestNetworkContextEntered:
+    def test_correct_choice_string_passed_to_parse_network_choice(self):
+
+        mock_ctx = _mock_network_ctx()
+
+        @network_context()
+        def handler(token: str) -> dict:
+            return {"token": token}
+
+        client = _make_app_and_client(handler)
+
+        with patch("ape.networks") as mock_networks:
+            mock_networks.parse_network_choice.return_value = mock_ctx
+            client.get("/test?token=USDC&ecosystem=ethereum&network=mainnet")
+
+        mock_networks.parse_network_choice.assert_called_once_with("ethereum:mainnet")
+
+    def test_different_ecosystem_and_network(self):
+
+        mock_ctx = _mock_network_ctx()
+
+        @network_context()
+        def handler() -> dict:
+            return {}
+
+        client = _make_app_and_client(handler)
+
+        with patch("ape.networks") as mock_networks:
+            mock_networks.parse_network_choice.return_value = mock_ctx
+            client.get("/test?ecosystem=polygon&network=mainnet")
+
+        mock_networks.parse_network_choice.assert_called_once_with("polygon:mainnet")
+
+
+# ---------------------------------------------------------------------------
+# Test 3 — kwargs passed to original function
+# ---------------------------------------------------------------------------
+
+
+class TestKwargsBehaviour:
+    def test_ecosystem_network_stripped_when_not_in_original_signature(self):
+
+        received: dict = {}
+
+        @network_context()
+        def handler(token: str) -> dict:
+            received.update({"token": token})
+            return received
+
+        client = _make_app_and_client(handler)
+
+        with patch("ape.networks") as mock_networks:
+            mock_networks.parse_network_choice.return_value = _mock_network_ctx()
+            response = client.get("/test?token=USDC&ecosystem=ethereum&network=mainnet")
+
+        assert response.status_code == 200
+        assert "ecosystem" not in received
+        assert "network" not in received
+        assert received["token"] == "USDC"
+
+    def test_ecosystem_network_passed_through_when_in_original_signature(self):
+
+        received: dict = {}
+
+        @network_context()
+        def handler(token: str, ecosystem: str, network: str) -> dict:
+            received.update({"token": token, "ecosystem": ecosystem, "network": network})
+            return received
+
+        client = _make_app_and_client(handler)
+
+        with patch("ape.networks") as mock_networks:
+            mock_networks.parse_network_choice.return_value = _mock_network_ctx()
+            response = client.get("/test?token=USDC&ecosystem=ethereum&network=mainnet")
+
+        assert response.status_code == 200
+        assert received == {"token": "USDC", "ecosystem": "ethereum", "network": "mainnet"}
+
+
+# ---------------------------------------------------------------------------
+# Test 4 — functools.wraps preserves original function metadata
+# ---------------------------------------------------------------------------
+
+
+class TestWrapsMetadata:
+    def test_function_name_preserved(self):
+        @network_context()
+        def my_special_handler() -> dict:
+            return {}
+
+        assert my_special_handler.__name__ == "my_special_handler"
+
+    def test_docstring_preserved(self):
+        @network_context()
+        def handler() -> dict:
+            """My docstring."""
+            return {}
+
+        assert handler.__doc__ == "My docstring."