Question - catch-all method handler

[nethi]

Hi,

With fastapi-jsonrpc, is it possible to have a catch-all method handler that gets invoked if no explicit method routers are registered ?

I am thinking of a use case where methods are dynamic where a pattern in method name indicates some namespace and what function logic to trigger

thanks Ramesh

[spumer]

You can inherit from EntrypointRoute class and override EntrypointRoute.handle_req method.

If i understand your question correctly you can do it like that

class EntrypointDynamicRouter(fastapi_jsonrpc.EntrypointRoute):
    def handle_req(self, *args, ctx, **kwargs):
        if not self.entrypoint.routers:
            route = self._here_your_function_to_dynamic_resolve_and_get_right_route(*args, **kwargs)
            ctx.method_route = route
            return await route.handle_req(...)

        # default resolver
        return super().handle_req(*args, ctx=ctx, **kwargs) 

class DynamicEntrypoint(fastapi_jsonrpc.Entrypoint):
    entrypoint_route_class = EntrypointDynamicRouter

app = fastapi_jsonrpc.API()

dynamic_api_v1 = DynamicEntrypoint('/api/v1/jsonrpc')

app.bind_entrypoint(dynamic_api_v1)

[spumer]

Also handling mechanism can be reworked to split method resolver and method running mechanisms.

It's not a problem, if it's helps solve your problem

[nethi]

Thank you. let me try and get back

On Thu, Oct 31, 2024 at 2:48 PM spumer @.***> wrote:

Also handling mechanism can be reworked to split method resolver and method running mechanisms

— Reply to this email directly, view it on GitHub https://github.com/smagafurov/fastapi-jsonrpc/issues/74#issuecomment-2449393177, or unsubscribe https://github.com/notifications/unsubscribe-auth/AABJSC5NX6JZQAK3MP72YR3Z6HYYBAVCNFSM6AAAAABQ5Z4UQKVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDINBZGM4TGMJXG4 . You are receiving this because you authored the thread.Message ID: @.***>

[nethi]

@spumer I followed your example and got a hello world to work. Thank you. one question. In this special case, I would like to make some data from "dynamic" context (that could come from a database or an external service) to be made available to method implementation ("echo" in this case), without modifying method signature. I am thinking I will use ContextVar approach - do you think it will work ? Or, if there is a better approach with fastapi or fastapi_jsonrpc ?

import fastapi_jsonrpc
from starlette.requests import Request
from starlette.responses import Response, JSONResponse

class EntrypointDynamicRouter(fastapi_jsonrpc.EntrypointRoute):
    def find_route(self, method: str):
        for route in self.entrypoint.routes:
            print(f"route.name: {route.name} {route.path}")
            if route.name == method:
                return route
        return None

    async def handle_req(        self,
        http_request: Request,
        background_tasks: fastapi_jsonrpc.BackgroundTasks,
        sub_response: Response,
        ctx: fastapi_jsonrpc.JsonRpcContext,
        dependency_cache: dict = None,
        shared_dependencies_error: fastapi_jsonrpc.BaseError = None):

        print(f"ctx.request.method: {ctx.request.method}")
        print(f"entrypoint.routers: {self.entrypoint.routes}")

        if ctx.request.method.startswith('dynamic_'):
            method_name = ctx.request.method[len('dynamic_'):]
            print(f"new method_name: {method_name}")
            route = self.find_route(method_name)
            if route:
                ctx.method_route = route
                return await route.handle_req(http_request, background_tasks, sub_response, 
                                ctx, 
                                dependency_cache, shared_dependencies_error)       
        # default resolver
        return await super().handle_req(http_request, background_tasks, sub_response, 
                                  ctx, 
                                  dependency_cache, shared_dependencies_error) 

class DynamicEntrypoint(fastapi_jsonrpc.Entrypoint):
    entrypoint_route_class = EntrypointDynamicRouter

app = fastapi_jsonrpc.API()

dynamic_api_v1 = DynamicEntrypoint('/api/v1/jsonrpc')

app.bind_entrypoint(dynamic_api_v1)

@dynamic_api_v1.method()
def echo(
    data: str) -> str:
    return data

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('jsonrpc-dynamic:app', port=8000, reload=True, access_log=False)

[nethi]

Btw, ContextVar approach seems to work fine.

[spumer]

Yes, or you can set attr for ctx: JsonRpcContext object and get it from fastapi_jsonrpc.get_jsonrpc_context() in any place.

But internally it works by ContextVar's too