stable/freqtrade/rpc/api_server2/webserver.py

from typing import Any, Dict, Optional
import uvicorn
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware

from freqtrade.rpc.rpc import RPC, RPCHandler

from .uvicorn_threaded import UvicornServer


class ApiServer(RPCHandler):

    _rpc: Optional[RPC] = None
    _config: Dict[str, Any] = {}

    def __init__(self, rpc: RPC, config: Dict[str, Any]) -> None:
        super().__init__(rpc, config)
        self._server = None

        ApiServer._rpc = rpc
        ApiServer._config = config

        self.app = FastAPI(title="Freqtrade API")
        self.configure_app(self.app, self._config)

        self.start_api()

    def cleanup(self) -> None:
        """ Cleanup pending module resources """
        if self._server:
            self._server.cleanup()

    def send_msg(self, msg: Dict[str, str]) -> None:
        pass

    def configure_app(self, app: FastAPI, config):
        from .api_v1 import router as api_v1
        from .api_v1 import router_public as api_v1_public
        from .auth import router_login, HTTPBasicOrJWTToken
        app.include_router(api_v1_public, prefix="/api/v1")

        app.include_router(api_v1, prefix="/api/v1",
                           dependencies=[Depends(HTTPBasicOrJWTToken())]
                           )
        app.include_router(router_login, prefix="/api/v1")

        app.add_middleware(
            CORSMiddleware,
            allow_origins=config['api_server'].get('CORS_origins', []),
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    def start_api(self):
        """
        Start API ... should be run in thread.
        """
        uvconfig = uvicorn.Config(self.app,
                                  port=self._config['api_server'].get('listen_port', 8080),
                                  host=self._config['api_server'].get(
                                      'listen_ip_address', '127.0.0.1'),
                                  access_log=True)
        self._server = UvicornServer(uvconfig)

        self._server.run_in_thread()