freqtrade_origin/freqtrade/rpc/api_server/deps.py

73 lines
2.0 KiB
Python
Raw Normal View History

2023-03-19 14:06:56 +00:00
from typing import Any, AsyncIterator, Dict, Optional
from uuid import uuid4
2021-01-02 11:54:40 +00:00
from fastapi import Depends, HTTPException
2022-01-22 06:11:59 +00:00
from freqtrade.constants import Config
from freqtrade.enums import RunMode
from freqtrade.persistence import Trade
from freqtrade.persistence.models import _request_id_ctx_var
from freqtrade.rpc.api_server.webserver_bgwork import ApiBG
2021-01-02 12:12:49 +00:00
from freqtrade.rpc.rpc import RPC, RPCException
2021-01-02 11:54:40 +00:00
from .webserver import ApiServer
2021-01-02 12:12:49 +00:00
def get_rpc_optional() -> Optional[RPC]:
if ApiServer._has_rpc:
return ApiServer._rpc
return None
2023-03-19 14:06:56 +00:00
async def get_rpc() -> Optional[AsyncIterator[RPC]]:
2021-01-02 12:12:49 +00:00
_rpc = get_rpc_optional()
if _rpc:
request_id = str(uuid4())
ctx_token = _request_id_ctx_var.set(request_id)
Trade.rollback()
try:
yield _rpc
finally:
Trade.session.remove()
_request_id_ctx_var.reset(ctx_token)
2021-01-02 12:12:49 +00:00
else:
raise RPCException('Bot is not in the correct state')
2021-01-02 11:54:40 +00:00
def get_config() -> Dict[str, Any]:
return ApiServer._config
2020-12-27 14:24:49 +00:00
2021-01-02 11:54:40 +00:00
def get_api_config() -> Dict[str, Any]:
2020-12-27 14:24:49 +00:00
return ApiServer._config['api_server']
2022-01-22 06:11:59 +00:00
def _generate_exchange_key(config: Config) -> str:
"""
Exchange key - used for caching the exchange object.
"""
return f"{config['exchange']['name']}_{config.get('trading_mode', 'spot')}"
2022-01-22 06:11:59 +00:00
def get_exchange(config=Depends(get_config)):
exchange_key = _generate_exchange_key(config)
if not (exchange := ApiBG.exchanges.get(exchange_key)):
2022-01-22 06:11:59 +00:00
from freqtrade.resolvers import ExchangeResolver
exchange = ExchangeResolver.load_exchange(
config, validate=False, load_leverage_tiers=False)
ApiBG.exchanges[exchange_key] = exchange
return exchange
2022-11-15 03:27:45 +00:00
def get_message_stream():
return ApiServer._message_stream
def is_webserver_mode(config=Depends(get_config)):
if config['runmode'] != RunMode.WEBSERVER:
raise HTTPException(status_code=503,
detail='Bot is not in the correct state.')
return None