Add data decoder endpoint

app/main.py

@@ -7,7 +7,7 @@
 from . import VERSION
 from .datasources.queue.exceptions import QueueProviderUnableToConnectException
 from .datasources.queue.queue_provider import QueueProvider
-from .routers import about, admin, contracts, default
+from .routers import about, admin, contracts, data_decoder, default
 from .services.abis import AbiService
 from .services.events import EventsService
 
@@ -58,5 +58,6 @@ async def lifespan(app: FastAPI):
 )
 api_v1_router.include_router(about.router)
 api_v1_router.include_router(contracts.router)
+api_v1_router.include_router(data_decoder.router)
 app.include_router(api_v1_router)
 app.include_router(default.router)

app/routers/data_decoder.py

@@ -0,0 +1,27 @@
+from fastapi import APIRouter, HTTPException
+
+from app.routers.models import DataDecodedPublic, DataDecoderInput
+from app.services.data_decoder import DataDecoded, get_data_decoder_service
+
+router = APIRouter(
+    prefix="/data-decoder",
+    tags=["data decoder"],
+)
+
+
+@router.post("", response_model=DataDecodedPublic)
+async def data_decoder(
+    input_data: DataDecoderInput,
+) -> DataDecoded:
+    data_decoder_service = await get_data_decoder_service()
+    # TODO: Add chainId to get_data_decoded
+    data_decoded = await data_decoder_service.get_data_decoded(
+        data=input_data.data, address=input_data.to
+    )
+
+    if data_decoded is None:
+        raise HTTPException(
+            status_code=404, detail="Cannot find function selector to decode data"
+        )
+
+    return data_decoded

app/routers/models.py

@@ -1,8 +1,14 @@
 from datetime import datetime
+from typing import Any, Union
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, Field, field_validator
 
-from safe_eth.eth.utils import ChecksumAddress, fast_to_checksum_address
+from eth_typing import HexStr
+from safe_eth.eth.utils import (
+    ChecksumAddress,
+    fast_is_checksum_address,
+    fast_to_checksum_address,
+)
 
 
 class About(BaseModel):
@@ -63,3 +69,43 @@ def convert_to_checksum_address(cls, address: bytes):
         if isinstance(address, bytes):
             return fast_to_checksum_address(address)
         return address
+
+
+class DataDecoderInput(BaseModel):
+    data: str = Field(
+        ..., pattern=r"^0x[0-9a-fA-F]*$", description="0x-prefixed hexadecimal string"
+    )
+    to: ChecksumAddress | None = Field(
+        None, pattern=r"^0x[0-9a-fA-F]{40}$", description="Optional to address"
+    )
+    chainId: int | None = Field(
+        None, gt=0, description="Optional Chain ID as a positive integer"
+    )
+
+    @field_validator("to")
+    def validate_checksum_address(cls, value):
+        if value and not fast_is_checksum_address(value):
+            raise ValueError("Address is not checksumed")
+        return value
+
+
+class ParameterDecodedPublic(BaseModel):
+    name: str
+    type: str
+    value: Any
+    value_decoded: (
+        Union[list["MultisendDecodedPublic"], "DataDecodedPublic", None] | None
+    ) = None
+
+
+class DataDecodedPublic(BaseModel):
+    method: str
+    parameters: list[ParameterDecodedPublic]
+
+
+class MultisendDecodedPublic(BaseModel):
+    operation: int
+    to: ChecksumAddress
+    value: str
+    data: HexStr | None = None
+    data_decoded: DataDecodedPublic | None = None

app/services/data_decoder.py

@@ -1,5 +1,4 @@
 import logging
-from functools import cache
 from typing import Any, AsyncIterator, NotRequired, TypedDict, Union, cast
 
 from async_lru import alru_cache
@@ -54,7 +53,7 @@ class MultisendDecoded(TypedDict):
     data_decoded: DataDecoded | None
 
 
-@cache
+@alru_cache
 @database_session
 async def get_data_decoder_service(session: AsyncSession) -> "DataDecoderService":
     data_decoder_service = DataDecoderService()

app/tests/routers/test_data_decoder.py

@@ -0,0 +1,73 @@
+import unittest
+
+from fastapi.testclient import TestClient
+
+from hexbytes import HexBytes
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from ...datasources.db.database import database_session
+from ...datasources.db.models import AbiSource
+from ...main import app
+from ...services.abis import AbiService
+
+
+class TestRouterAbout(unittest.TestCase):
+    client: TestClient
+
+    @classmethod
+    def setUpClass(cls):
+        cls.client = TestClient(app)
+
+    @database_session
+    async def test_view_data_decoder(self, session: AsyncSession):
+        # Add safe abis for testing
+        abi_service = AbiService()
+        safe_abis = abi_service.get_safe_abis()
+        abi_source, _ = await AbiSource.get_or_create(
+            session, "localstorage", "decoder-service"
+        )
+        await abi_service._store_abis_in_database(session, safe_abis, 100, abi_source)
+
+        add_owner_with_threshold_data = HexBytes(
+            "0x0d582f130000000000000000000000001b9a0da11a5cace4e7035993cbb2e4"
+            "b1b3b164cf000000000000000000000000000000000000000000000000000000"
+            "0000000001"
+        )
+
+        response = self.client.post(
+            "/api/v1/data-decoder/", json={"data": add_owner_with_threshold_data.hex()}
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.json(),
+            {
+                "method": "addOwnerWithThreshold",
+                "parameters": [
+                    {
+                        "name": "owner",
+                        "type": "address",
+                        "value": "0x1b9a0DA11a5caCE4e7035993Cbb2E4B1B3b164Cf",
+                        "value_decoded": None,
+                    },
+                    {
+                        "name": "_threshold",
+                        "type": "uint256",
+                        "value": "1",
+                        "value_decoded": None,
+                    },
+                ],
+            },
+        )
+
+        response = self.client.post("/api/v1/data-decoder/", json={"data": "0x123"})
+        self.assertEqual(response.status_code, 404)
+
+        # Test no checksumed address
+        response = self.client.post(
+            "/api/v1/data-decoder/",
+            json={
+                "data": add_owner_with_threshold_data.hex(),
+                "to": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
+            },
+        )
+        self.assertEqual(response.status_code, 422)