stable/freqtrade/rpc/api_server2/webserver.py

67 lines
2.1 KiB
Python
Raw Normal View History

2020-12-25 14:50:19 +00:00
from typing import Any, Dict, Optional
import uvicorn
2020-12-25 14:50:19 +00:00
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
2020-12-25 14:50:19 +00:00
from freqtrade.rpc.rpc import RPC, RPCHandler
from .uvicorn_threaded import UvicornServer
class ApiServer(RPCHandler):
2020-12-25 14:50:19 +00:00
_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
2020-12-25 14:50:19 +00:00
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
2020-12-25 14:50:19 +00:00
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")
2020-12-25 14:50:19 +00:00
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()