diff --git a/ytnoti/__init__.py b/ytnoti/__init__.py index 33d48ff..242ea61 100644 --- a/ytnoti/__init__.py +++ b/ytnoti/__init__.py @@ -81,6 +81,7 @@ def run(self, :param app: The FastAPI app to use. If not provided, a new app will be created. :param log_level: The log level to use for the uvicorn server. :param configs: Additional arguments to pass to the Config class of uvicorn. + :raises ValueError: If the given app instance has a route that conflicts with the notifier's routes. """ server = super()._get_server(host=host, port=port, app=app, log_level=log_level, **configs) @@ -182,6 +183,7 @@ async def serve(self, :param app: The FastAPI app instance to use. If not provided, a new instance will be created. :param configs: Additional arguments to pass to the Config class of uvicorn. + :raises ValueError: If the given app instance has a route that conflicts with the notifier's routes. :raises RuntimeError: If the method is not called from a running event loop. """ diff --git a/ytnoti/base.py b/ytnoti/base.py index 92e7ecd..dc6809b 100644 --- a/ytnoti/base.py +++ b/ytnoti/base.py @@ -17,6 +17,8 @@ from httpx import AsyncClient import xmltodict from fastapi import FastAPI, Request, Response, APIRouter +from fastapi.routing import APIRoute +from starlette.routing import Route from uvicorn import Config, Server from pyngrok import ngrok from pyexpat import ExpatError @@ -296,6 +298,7 @@ def _get_server(self, :param log_level: The log level to use for the uvicorn server. :param configs: Additional arguments to pass to the server configuration. :return: The server instance. + :raises ValueError: If the given app instance has a route that conflicts with the notifier's routes. """ self._config.host = host @@ -307,6 +310,11 @@ def _get_server(self, self.__logger.info("Callback URL: %s", self._config.callback_url) + endpoint = urlparse(self._config.callback_url).path or "/" + + if any(isinstance(route, (APIRoute, Route)) and route.path == endpoint for route in app.routes): + raise ValueError(f"Endpoint {endpoint} is reserved for {__package__} so it cannot be used by the app") + self._config.app.include_router(self._get_router()) async def on_ready():