import json
import logging
import os
import urllib3.util
from dotenv import dotenv_values
from lime_trader.clients.account_client import AccountClient
from lime_trader.constants.config import DEFAULT_TIMEOUT
from lime_trader.converters.cattr_converter import CAttrConverter
from lime_trader.models.accounts import Credentials
from lime_trader.api.api_client import ApiClient
from lime_trader.api.authenticated_api_client import AuthenticatedApiClient
from lime_trader.clients.market_data_client import MarketDataClient
from lime_trader.clients.trading_client import TradingClient
from lime_trader.models.token_storage import ApiTokenStorage
from lime_trader.utils.logging_utils import get_stdout_logger
[docs]
class LimeClient:
"""
Main client that should be used for all communication with the API.
Attributes:
account: Account client
market: Market client
trading: Trading client
"""
def __init__(self, base_url: str, credentials: Credentials, logger: logging.Logger | None = None,
auth_url: str | None = None, timeout: int = DEFAULT_TIMEOUT):
"""
Args:
base_url: Base url of the API, this will be prepended to each url
credentials: Credentials object used for authenticating in the API
logger: Logger to be used
auth_url: URL of the authorization server. Only requests related to authentication/authorization
will use this url
timeout: Request timeout
"""
self._base_url = urllib3.util.parse_url(base_url)
self._auth_url = urllib3.util.parse_url(auth_url) if auth_url is not None else self._base_url
self._api_use_https = self._base_url.url.startswith("https://")
self._api_use_wss = self._api_use_https
self._auth_use_https = self._auth_url.url.startswith("https://")
self._auth_use_wss = self._auth_use_https
self._credentials = credentials
self._logger = logger or get_stdout_logger()
self._converter = CAttrConverter()
self._api_client = ApiClient(base_url=self._base_url, converter=self._converter, use_https=self._api_use_https,
use_wss=self._api_use_wss, logger=self._logger, timeout=timeout)
self._auth_api_client = ApiClient(base_url=self._auth_url, converter=self._converter,
use_https=self._auth_use_https,
use_wss=self._auth_use_wss, logger=self._logger, timeout=timeout)
self._token_storage = ApiTokenStorage(credentials=credentials, api_client=self._auth_api_client)
self._authenticated_api_client = AuthenticatedApiClient(api_client=self._api_client,
token_storage=self._token_storage,
credentials=self._credentials, logger=self._logger)
# instantiate clients
self.account = AccountClient(api_client=self._authenticated_api_client, logger=self._logger)
self.trading = TradingClient(api_client=self._authenticated_api_client, logger=self._logger)
self.market = MarketDataClient(api_client=self._authenticated_api_client, logger=self._logger)
[docs]
@classmethod
def from_json(cls, json_str: str, logger: logging.Logger | None = None) -> 'LimeClient':
"""
Instantiates client from json string.
Required JSON keys are:
- username
- password
- client_id
- client_secret
- grant_type
- base_url
- auth_url
Args:
json_str: JSON string
logger: Logger to use
Returns:
Configured LimeClient
"""
data = json.loads(json_str)
return cls.from_dict(data=data, logger=logger)
[docs]
@classmethod
def from_file(cls, file_path: str, logger: logging.Logger | None = None) -> 'LimeClient':
"""
Instantiates client from file. Contents of the file needs to be json
Required JSON keys are:
- username
- password
- client_id
- client_secret
- grant_type
- base_url
- auth_url
Args:
file_path: Path to the file
logger: Logger to use
Returns:
Configured LimeClient
"""
with open(file=file_path, mode="r") as f:
content = f.read()
return cls.from_json(json_str=content, logger=logger)
[docs]
@classmethod
def from_dict(cls, data: dict[str, str], logger: logging.Logger | None = None) -> 'LimeClient':
"""
Instantiates client from dictionary.
Dictionary must have required keys:
- username
- password
- client_id
- client_secret
- grant_type
- base_url
- auth_url
Args:
data: Dictionary with client data and credentials
logger: Logger to use
Returns:
Configured LimeClient
"""
credentials = Credentials(username=data["username"],
password=data["password"],
client_id=data["client_id"],
client_secret=data["client_secret"],
grant_type=data["grant_type"])
client = cls(base_url=data["base_url"],
credentials=credentials,
auth_url=data["auth_url"],
logger=logger)
return client
[docs]
@classmethod
def from_env(cls, logger: logging.Logger | None = None) -> 'LimeClient':
"""
Configures client by getting credentials and configuration from system environment.
Required environment variables are:
- LIME_SDK_USERNAME
- LIME_SDK_PASSWORD
- LIME_SDK_CLIENT_ID
- LIME_SDK_CLIENT_SECRET
- LIME_SDK_GRANT_TYPE
- LIME_SDK_BASE_URL
- LIME_SDK_AUTH_URL
Args:
logger: Logger to use
Returns:
Configured LimeClient
"""
credentials = {
"username": os.environ.get("LIME_SDK_USERNAME"),
"password": os.environ.get("LIME_SDK_PASSWORD"),
"client_id": os.environ.get("LIME_SDK_CLIENT_ID"),
"client_secret": os.environ.get("LIME_SDK_CLIENT_SECRET"),
"grant_type": os.environ.get("LIME_SDK_GRANT_TYPE"),
"base_url": os.environ.get("LIME_SDK_BASE_URL"),
"auth_url": os.environ.get("LIME_SDK_AUTH_URL"),
}
return cls.from_dict(credentials, logger)
[docs]
@classmethod
def from_env_file(cls, file_path: str, logger: logging.Logger | None = None) -> 'LimeClient':
"""
Configures client by getting credentials and configuration from key/value file, usually named `.env`.
Required keys are:
- LIME_SDK_USERNAME
- LIME_SDK_PASSWORD
- LIME_SDK_CLIENT_ID
- LIME_SDK_CLIENT_SECRET
- LIME_SDK_GRANT_TYPE
- LIME_SDK_BASE_URL
- LIME_SDK_AUTH_URL
Args:
file_path: File name from which to load config
logger: Logger to use
Returns:
Configured LimeClient
"""
config = dotenv_values(file_path)
credentials = {
"username": config.get("LIME_SDK_USERNAME"),
"password": config.get("LIME_SDK_PASSWORD"),
"client_id": config.get("LIME_SDK_CLIENT_ID"),
"client_secret": config.get("LIME_SDK_CLIENT_SECRET"),
"grant_type": config.get("LIME_SDK_GRANT_TYPE"),
"base_url": config.get("LIME_SDK_BASE_URL"),
"auth_url": config.get("LIME_SDK_AUTH_URL"),
}
return cls.from_dict(credentials, logger)