import logging from ipaddress import ip_address from typing import Any, Optional import orjson import uvicorn from fastapi import Depends, FastAPI from fastapi.middleware.cors import CORSMiddleware from starlette.responses import JSONResponse from freqtrade.configuration import running_in_docker from freqtrade.constants import Config from freqtrade.exceptions import OperationalException from freqtrade.rpc.api_server.uvicorn_threaded import UvicornServer from freqtrade.rpc.api_server.ws.message_stream import MessageStream from freqtrade.rpc.rpc import RPC, RPCException, RPCHandler from freqtrade.rpc.rpc_types import RPCSendMsg logger = logging.getLogger(__name__) class FTJSONResponse(JSONResponse): media_type = "application/json" def render(self, content: Any) -> bytes: """ Use rapidjson for responses Handles NaN and Inf / -Inf in a javascript way by default. """ return orjson.dumps(content, option=orjson.OPT_SERIALIZE_NUMPY) class ApiServer(RPCHandler): __instance = None __initialized = False _rpc: RPC _has_rpc: bool = False _config: Config = {} # websocket message stuff _message_stream: Optional[MessageStream] = None def __new__(cls, *args, **kwargs): """ This class is a singleton. We'll only have one instance of it around. """ if ApiServer.__instance is None: ApiServer.__instance = object.__new__(cls) ApiServer.__initialized = False return ApiServer.__instance def __init__(self, config: Config, standalone: bool = False) -> None: ApiServer._config = config if self.__initialized and (standalone or self._standalone): return self._standalone: bool = standalone self._server = None ApiServer.__initialized = True api_config = self._config["api_server"] self.app = FastAPI( title="Freqtrade API", docs_url="/docs" if api_config.get("enable_openapi", False) else None, redoc_url=None, default_response_class=FTJSONResponse, ) self.configure_app(self.app, self._config) self.start_api() def add_rpc_handler(self, rpc: RPC): """ Attach rpc handler """ if not ApiServer._has_rpc: ApiServer._rpc = rpc ApiServer._has_rpc = True else: # This should not happen assuming we didn't mess up. raise OperationalException("RPC Handler already attached.") def cleanup(self) -> None: """Cleanup pending module resources""" ApiServer._has_rpc = False del ApiServer._rpc if self._server and not self._standalone: logger.info("Stopping API Server") # self._server.force_exit, self._server.should_exit = True, True self._server.cleanup() @classmethod def shutdown(cls): cls.__initialized = False del cls.__instance cls.__instance = None cls._has_rpc = False cls._rpc = None def send_msg(self, msg: RPCSendMsg) -> None: """ Publish the message to the message stream """ if ApiServer._message_stream: ApiServer._message_stream.publish(msg) def handle_rpc_exception(self, request, exc): logger.error(f"API Error calling: {exc}") return JSONResponse( status_code=502, content={"error": f"Error querying {request.url.path}: {exc.message}"} ) def configure_app(self, app: FastAPI, config): from freqtrade.rpc.api_server.api_auth import http_basic_or_jwt_token, router_login from freqtrade.rpc.api_server.api_background_tasks import router as api_bg_tasks from freqtrade.rpc.api_server.api_backtest import router as api_backtest from freqtrade.rpc.api_server.api_v1 import router as api_v1 from freqtrade.rpc.api_server.api_v1 import router_public as api_v1_public from freqtrade.rpc.api_server.api_ws import router as ws_router from freqtrade.rpc.api_server.deps import is_webserver_mode from freqtrade.rpc.api_server.web_ui import router_ui app.include_router(api_v1_public, prefix="/api/v1") app.include_router(router_login, prefix="/api/v1", tags=["auth"]) app.include_router( api_v1, prefix="/api/v1", dependencies=[Depends(http_basic_or_jwt_token)], ) app.include_router( api_backtest, prefix="/api/v1", dependencies=[Depends(http_basic_or_jwt_token), Depends(is_webserver_mode)], ) app.include_router( api_bg_tasks, prefix="/api/v1", dependencies=[Depends(http_basic_or_jwt_token), Depends(is_webserver_mode)], ) app.include_router(ws_router, prefix="/api/v1") # UI Router MUST be last! app.include_router(router_ui, prefix="") app.add_middleware( CORSMiddleware, allow_origins=config["api_server"].get("CORS_origins", []), allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.add_exception_handler(RPCException, self.handle_rpc_exception) app.add_event_handler(event_type="startup", func=self._api_startup_event) app.add_event_handler(event_type="shutdown", func=self._api_shutdown_event) async def _api_startup_event(self): """ Creates the MessageStream class on startup so it has access to the same event loop as uvicorn """ if not ApiServer._message_stream: ApiServer._message_stream = MessageStream() async def _api_shutdown_event(self): """ Removes the MessageStream class on shutdown """ if ApiServer._message_stream: ApiServer._message_stream = None def start_api(self): """ Start API ... should be run in thread. """ rest_ip = self._config["api_server"]["listen_ip_address"] rest_port = self._config["api_server"]["listen_port"] logger.info(f"Starting HTTP Server at {rest_ip}:{rest_port}") if not ip_address(rest_ip).is_loopback and not running_in_docker(): logger.warning("SECURITY WARNING - Local Rest Server listening to external connections") logger.warning( "SECURITY WARNING - This is insecure please set to your loopback," "e.g 127.0.0.1 in config.json" ) if not self._config["api_server"].get("password"): logger.warning( "SECURITY WARNING - No password for local REST Server defined. " "Please make sure that this is intentional!" ) if self._config["api_server"].get("jwt_secret_key", "super-secret") in ( "super-secret, somethingrandom" ): logger.warning( "SECURITY WARNING - `jwt_secret_key` seems to be default." "Others may be able to log into your bot." ) logger.info("Starting Local Rest Server.") verbosity = self._config["api_server"].get("verbosity", "error") uvconfig = uvicorn.Config( self.app, port=rest_port, host=rest_ip, use_colors=False, log_config=None, access_log=True if verbosity != "error" else False, ws_ping_interval=None, # We do this explicitly ourselves ) try: self._server = UvicornServer(uvconfig) if self._standalone: self._server.run() else: self._server.run_in_thread() except Exception: logger.exception("Api server failed to start.")