jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-11-03 07:51:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

first round, refactor registry, executor, map

Browse Source
This commit is contained in:
Diogo Sousa
2023-07-27 14:31:29 +01:00
parent db85048467
commit 027bab20c4
13 changed files with 321 additions and 659 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
openbb_sdk/providers/benzinga/openbb_benzinga/models/global_news.py
View File

@@ -2,7 +2,7 @@
from datetime import datetime
from typing import Dict, List, Literal, Optional
from typing import Any, Dict, List, Literal, Optional
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.helpers import data_transformer, get_querystring
@@ -104,19 +104,15 @@ class BenzingaGlobalNewsFetcher(
]
):
@staticmethod
def transform_query(
query: GlobalNewsQueryParams, extra_params: Optional[Dict] = None
) -> BenzingaGlobalNewsQueryParams:
return BenzingaGlobalNewsQueryParams(
page=query.page, **extra_params if extra_params else {}
)
def transform_query(params: Dict[str, Any]) -> BenzingaGlobalNewsQueryParams:
return BenzingaGlobalNewsQueryParams(**params)
@staticmethod
def extract_data(
query: BenzingaGlobalNewsQueryParams, credentials: Optional[Dict[str, str]]
) -> List[BenzingaGlobalNewsData]:
if credentials:
api_key = credentials.get("benzinga_api_key")
print(query)
api_key = credentials.get("benzinga_api_key") or ""
base_url = "https://api.benzinga.com/api/v2/news"
querystring = get_querystring(query.dict(by_alias=True), [])
@@ -131,6 +127,7 @@ class BenzingaGlobalNewsFetcher(
@staticmethod
def transform_data(
data: List[BenzingaGlobalNewsData],
) -> List[GlobalNewsData]:
) -> List[BenzingaGlobalNewsData]:
return data
processors = {"image": lambda x: "" if x == [] else x[0].url}
return data_transformer(data, GlobalNewsData, processors)
return data_transformer(data, BenzingaGlobalNewsData, processors)

13
openbb_sdk/providers/benzinga/openbb_benzinga/models/stock_news.py
View File

@@ -2,7 +2,7 @@
from datetime import datetime
from typing import Dict, List, Literal, Optional
from typing import Any, Dict, List, Literal, Optional
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.helpers import data_transformer, get_querystring
@@ -100,15 +100,8 @@ class BenzingaStockNewsFetcher(
]
):
@staticmethod
def transform_query(
query: StockNewsQueryParams, extra_params: Optional[Dict] = None
) -> BenzingaStockNewsQueryParams:
return BenzingaStockNewsQueryParams(
page=query.page,
tickers=query.symbols, # type: ignore
pageSize=query.limit,
**extra_params if extra_params else {},
)
def transform_query(params: Dict[str, Any]) -> BenzingaStockNewsQueryParams:
return BenzingaStockNewsQueryParams(**params)
@staticmethod
def extract_data(

16
openbb_sdk/providers/fred/openbb_fred/models/cpi.py
View File

@@ -1,7 +1,7 @@
"""FRED CPI Fetcher."""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from openbb_fred.utils.fred_base import Fred
from openbb_fred.utils.fred_helpers import all_cpi_options
@@ -19,18 +19,8 @@ class FREDCPIData(CPIData):
class FREDCPIFetcher(Fetcher[CPIQueryParams, CPIData, FREDCPIQueryParams, FREDCPIData]):
@staticmethod
def transform_query(
query: CPIQueryParams, extra_params: Optional[Dict] = None
) -> FREDCPIQueryParams:
return FREDCPIQueryParams(
countries=query.countries,
units=query.units,
frequency=query.frequency,
harmonized=query.harmonized,
start_date=query.start_date,
end_date=query.end_date,
**extra_params if extra_params else {},
)
def transform_query(params: Dict[str, Any]) -> FREDCPIQueryParams:
return FREDCPIQueryParams(**params)
@staticmethod
def extract_data(

17
openbb_sdk/sdk/core/openbb_core/app/model/credentials.py
View File

@@ -1,4 +1,5 @@
from pydantic import create_model
from typing import Dict, List, Optional, Tuple
from pydantic import create_model, Extra
from openbb_core.app.provider_interface import get_provider_interface
@@ -9,12 +10,24 @@ from openbb_core.app.provider_interface import get_provider_interface
class Config:
validate_assignment = True
# extra = Extra.forbid
def format_map(
required_credentials: List[str],
) -> Dict[str, Tuple[Optional[str], None]]:
"""Format credentials map to be used in the Credentials model"""
formatted: Dict[str, Tuple[Optional[str], None]] = {}
for c in required_credentials:
formatted[c] = (Optional[str], None)
return formatted
Credentials = create_model( # type: ignore
"Credentials",
__config__=Config,
**get_provider_interface().required_credentials,
**format_map(get_provider_interface().required_credentials),
)

126
openbb_sdk/sdk/core/openbb_core/app/provider_interface.py
View File

@@ -2,13 +2,10 @@ from dataclasses import dataclass, make_dataclass
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
from fastapi import Query
from openbb_provider.map import build_credentials_mapping, build_model_mapping
from openbb_provider.registry import (
ProviderRegistry,
build_provider_registry,
extensions_dict__,
)
from openbb_provider.query_executor import QueryExecutor
from openbb_provider.registry_map import RegistryMap
from pydantic import BaseConfig, BaseModel, Extra, Field, create_model
from pydantic.fields import ModelField
@dataclass
@@ -73,55 +70,51 @@ class ProviderInterface:
Build provider registry
"""
def __init__(self) -> None:
self.__map = build_model_mapping()
self.__model_providers_map = self.__generate_model_providers_dc()
self.__params = self.__generate_params_dc()
self.__data = self.__generate_data_dc()
self.__merged_data = self.__merge_data_dc(self.__data)
self.__required_credentials = self.__get_required_credentials()
def __init__(self, registry_map: Optional[RegistryMap] = None) -> None:
"""Initialize provider interface."""
self._registry_map = registry_map or RegistryMap()
self._map = self._registry_map.map
# self._map = build_model_mapping()
self._model_providers_map = self._generate_model_providers_dc()
self._params = self._generate_params_dc()
self._data = self._generate_data_dc()
self._merged_data = self._merge_data_dc(self._data)
self._required_credentials = self._get_required_credentials()
@property
def map(self) -> Dict[str, Dict[str, Dict[str, Any]]]:
"""Dictionary of provider information."""
return self.__map
return self._map
@property
def required_credentials(self) -> Dict[str, Tuple[Optional[str], None]]:
def required_credentials(self) -> Dict[str, Dict[str, str]]:
"""Dictionary of required credentials by provider."""
return self.__required_credentials
return self._required_credentials
@property
def model_providers(self) -> Dict[str, ProviderChoices]:
"""Dictionary of provider choices by model."""
return self.__model_providers_map
return self._model_providers_map
@property
def params(self) -> Dict[str, Dict[str, Union[StandardParams, ExtraParams]]]:
"""Dictionary of params by model."""
return self.__params
return self._params
@property
def data(self) -> Dict[str, Dict[str, Union[StandardData, ExtraData]]]:
"""Dictionary of data by model."""
return self.__data
return self._data
@property
def merged_data(self) -> Dict[str, BaseModel]:
"""Dictionary of data by model merged."""
return self.__merged_data
return self._merged_data
@property
def providers_literal(self) -> type:
"""Literal of provider names."""
providers = []
for _, provider in self.map.items():
providers.extend(list(provider.keys()))
providers = list(set(providers))
if "openbb" in providers:
providers.remove("openbb")
return Literal[tuple(providers)] # type: ignore
return Literal[tuple(self._registry_map.available_providers)] # type: ignore
@property
def provider_choices(self) -> type:
@@ -135,25 +128,18 @@ class ProviderInterface:
@property
def models(self) -> List[str]:
"""List of model names."""
return list(self.__map.keys())
return self._registry_map.models
def build_registry(self) -> ProviderRegistry:
"""Build provider registry."""
return build_provider_registry(extensions_dict__)
def create_executor(self) -> QueryExecutor:
"""Get query executor."""
return QueryExecutor(self._registry_map.registry)
@staticmethod
def __get_required_credentials() -> Dict[str, Tuple[Optional[str], None]]:
def _get_required_credentials(self) -> List[str]:
"""Get required credentials."""
required_credentials: Dict[str, Tuple[Optional[str], None]] = {}
credentials_mapping = build_credentials_mapping()
for credentials in credentials_mapping.values():
for c in credentials:
required_credentials[c] = (Optional[str], None)
return required_credentials
return self._registry_map.required_credentials
@staticmethod
def __merge_fields(
def _merge_fields(
current: DataclassField, incoming: DataclassField, query: bool = False
) -> DataclassField:
current_name = current.name
@@ -181,17 +167,17 @@ class ProviderInterface:
)
@staticmethod
def __create_field(
def _create_field(
name: str,
field: Dict[str, Any],
field: ModelField,
provider_name: Optional[str] = None,
query: bool = False,
) -> DataclassField:
new_name = name.replace(".", "_")
type_ = field["type"]
description = field["description"]
type_ = field.type_
description = field.field_info.description
default = ... if field["required"] else field["default"]
default = ... if field.required else field.default
if query:
# We need to use query if we want the field description to show up in the
# swagger, it's a fastapi limitation
@@ -206,7 +192,7 @@ class ProviderInterface:
return DataclassField(new_name, type_, default)
@classmethod
def __extract_params(
def _extract_params(
cls,
providers: Any,
) -> Tuple[Dict[str, Tuple[str, Any, Any]], Dict[str, Tuple[str, Any, Any]]]:
@@ -215,8 +201,8 @@ class ProviderInterface:
for provider_name, model_details in providers.items():
if provider_name == "openbb":
for name, field in model_details["QueryParams"]["fields"].items():
incoming = cls.__create_field(name, field, query=True)
for name, field in model_details["QueryParams"].items():
incoming = cls._create_field(name, field, query=True)
standard[incoming.name] = (
incoming.name,
@@ -224,15 +210,15 @@ class ProviderInterface:
incoming.default,
)
else:
for name, field in model_details["QueryParams"]["fields"].items():
if name not in providers["openbb"]["QueryParams"]["fields"]:
incoming = cls.__create_field(
for name, field in model_details["QueryParams"].items():
if name not in providers["openbb"]["QueryParams"]:
incoming = cls._create_field(
name, field, provider_name, query=True
)
if incoming.name in extra:
current = DataclassField(*extra[incoming.name])
updated = cls.__merge_fields(current, incoming, query=True)
updated = cls._merge_fields(current, incoming, query=True)
else:
updated = incoming
@@ -252,7 +238,7 @@ class ProviderInterface:
return standard, extra
@classmethod
def __extract_data(
def _extract_data(
cls,
providers: Any,
) -> Tuple[Dict[str, Tuple[str, Any, Any]], Dict[str, Tuple[str, Any, Any]]]:
@@ -261,8 +247,8 @@ class ProviderInterface:
for provider_name, model_details in providers.items():
if provider_name == "openbb":
for name, field in model_details["Data"]["fields"].items():
incoming = cls.__create_field(name, field, "standard")
for name, field in model_details["Data"].items():
incoming = cls._create_field(name, field, "openbb")
standard[incoming.name] = (
incoming.name,
@@ -270,13 +256,13 @@ class ProviderInterface:
incoming.default,
)
else:
for name, field in model_details["Data"]["fields"].items():
if name not in providers["openbb"]["Data"]["fields"]:
incoming = cls.__create_field(name, field, provider_name)
for name, field in model_details["Data"].items():
if name not in providers["openbb"]["Data"]:
incoming = cls._create_field(name, field, provider_name)
if incoming.name in extra:
current = DataclassField(*extra[incoming.name])
updated = cls.__merge_fields(current, incoming)
updated = cls._merge_fields(current, incoming)
else:
updated = incoming
@@ -288,7 +274,7 @@ class ProviderInterface:
return standard, extra
def __generate_params_dc(
def _generate_params_dc(
self,
) -> Dict[str, Dict[str, Union[StandardParams, ExtraParams]]]:
"""Generate dataclasses for params.
@@ -312,8 +298,8 @@ class ProviderInterface:
result: Dict = {}
# TODO: Consider multiprocessing this loop to speed startup
for model_name, providers in self.__map.items():
standard, extra = self.__extract_params(providers)
for model_name, providers in self._map.items():
standard, extra = self._extract_params(providers)
result[model_name] = {
"standard": make_dataclass( # type: ignore
@@ -329,7 +315,7 @@ class ProviderInterface:
}
return result
def __generate_model_providers_dc(self) -> Dict[str, ProviderChoices]:
def _generate_model_providers_dc(self) -> Dict[str, ProviderChoices]:
"""Generate dataclasses for provider choices by model.
This creates a dictionary that maps model names to dataclasses that can be used
@@ -343,7 +329,7 @@ class ProviderInterface:
"""
result: Dict = {}
for model_name, providers in self.__map.items():
for model_name, providers in self._map.items():
choices = list(providers.keys())
if "openbb" in choices:
choices.remove("openbb")
@@ -356,7 +342,7 @@ class ProviderInterface:
return result
def __generate_data_dc(
def _generate_data_dc(
self,
) -> Dict[str, Dict[str, Union[StandardData, ExtraData]]]:
"""Generate dataclasses for data.
@@ -377,8 +363,8 @@ class ProviderInterface:
result: Dict = {}
# TODO: Consider multiprocessing this loop to speed startup
for model_name, providers in self.__map.items():
standard, extra = self.__extract_data(providers)
for model_name, providers in self._map.items():
standard, extra = self._extract_data(providers)
result[model_name] = {
"standard": make_dataclass( # type: ignore
cls_name=model_name,
@@ -394,7 +380,7 @@ class ProviderInterface:
return result
def __merge_data_dc(
def _merge_data_dc(
self, data: Dict[str, Dict[str, Union[StandardData, ExtraData]]]
) -> Dict[str, BaseModel]:
"""Merge standard data with extra data into a single BaseModel."""

27
openbb_sdk/sdk/core/openbb_core/app/query.py
View File

@@ -60,28 +60,19 @@ class Query:
return filtered
def to_query_params(self, standard_params: StandardParams) -> StandardParams:
"""Convert standard params to QueryParams like class."""
standard_params.__name__ = self.name + "QueryParams" # type: ignore
return standard_params
def execute(self) -> BaseModel:
def execute(self) -> Any:
"""Execute the query."""
# TODO: Understand if we really need to create the registry in every call
registry = get_provider_interface().build_registry()
creds = self.cc.user_settings.credentials.dict()
query_params = self.to_query_params(self.standard_params)
filtered = (
executor = get_provider_interface().create_executor()
standard_dict = asdict(self.standard_params)
extra_dict = (
self.filter_extra_params(self.extra_params, self.provider)
if self.extra_params
else None
else {}
)
return registry.fetch(
return executor.execute(
provider_name=self.provider,
query_params=query_params,
extra_params=filtered,
credentials=creds,
model_name=self.name,
params={**standard_dict, **extra_dict},
credentials=self.cc.user_settings.credentials.dict(),
)

8
openbb_sdk/sdk/core/openbb_core/app/static/package_builder.py
View File

@@ -565,10 +565,10 @@ class MethodDefinition:
sig = signature(func)
parameter_map = dict(sig.parameters)
if model_name:
func = DocstringGenerator.generate_command_docstring(
func=func, model_name=model_name
)
# if model_name:
# func = DocstringGenerator.generate_command_docstring(
# func=func, model_name=model_name
# )
code = cls.build_command_method_signature(
func_name=func_name,

52
openbb_sdk/sdk/provider/openbb_provider/abstract/fetcher.py
View File

@@ -1,7 +1,7 @@
"""Abstract class for the fetcher."""
from typing import Dict, Generic, List, Optional, TypeVar, Union
from typing import Any, Dict, Generic, List, Optional, TypeVar, Union
from openbb_provider.abstract.data import Data, QueryParams
@@ -24,54 +24,54 @@ class Fetcher(
"""Abstract class for the fetcher."""
@staticmethod
def transform_query(
query: QueryParamsType,
extra_params: Optional[Dict] = None,
) -> ProviderQueryParamsType:
"""Transform the standard query to the provider-specific query."""
raise NotImplementedError("Method `transform_query` not implemented.")
def transform_query(params: Dict[str, Any]) -> ProviderQueryParamsType:
"""Transform the params to the provider-specific query."""
raise NotImplementedError
@staticmethod
def extract_data(
query: ProviderQueryParamsType, credentials: Optional[Dict[str, str]]
) -> ReturnType:
"""Extract the data from the provider."""
raise NotImplementedError("Method `extract_data` not implemented.")
raise NotImplementedError
@staticmethod
def transform_data(data: ReturnType) -> UDataType:
"""Transform the provider-specific data to the standard data."""
raise NotImplementedError("Method `transform_query` not implemented.")
raise NotImplementedError
@classmethod
def fetch_data(
cls,
query: QueryParamsType,
extra_params: Optional[Dict],
params: Dict[str, Any],
credentials: Optional[Dict[str, str]] = None,
) -> UDataType:
"""Fetch data from a provider by using the OpenBB standard."""
query = cls.transform_query(query=query, extra_params=extra_params)
data = cls.extract_data(query=query, credentials=credentials)
# data = cls.transform_data(data=data)
provider_query = cls.transform_query(params=params)
provider_data = cls.extract_data(query=provider_query, credentials=credentials)
data = cls.transform_data(data=provider_data)
return data
@classmethod
def get_query_type(cls):
@property
def query_params_type(self):
"""Get the type of the query."""
return cls.transform_query.__annotations__["query"]
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[0]
@classmethod
def get_data_type(cls):
@property
def data_type(self):
"""Get the type of the data."""
return cls.transform_data.__annotations__["return"]
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[1]
@classmethod
def get_provider_query_type(cls):
@property
def provider_query_params_type(self):
"""Get the type of the provider query."""
return cls.extract_data.__annotations__["query"]
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[2]
@classmethod
def get_provider_data_type(cls):
@property
def provider_data_type(self):
"""Get the type of the provider data."""
return cls.extract_data.__annotations__["return"]
# pylint: disable=E1101
return self.__orig_bases__[0].__args__[3]

32
openbb_sdk/sdk/provider/openbb_provider/abstract/provider.py
View File

@@ -8,8 +8,6 @@ from openbb_provider.abstract.fetcher import Fetcher
class Provider:
"""Abstract class for providers."""
QUERIES = ["get_query_type", "get_provider_query_type"]
def __init__(
self,
name: str,
@@ -17,28 +15,16 @@ class Provider:
required_credentials: Optional[List[str]],
fetcher_list: List[Fetcher],
) -> None:
"""Initialize the provider.
Parameters
----------
name : str
The name of the provider.
description : str
The description of the provider.
required_credentials : Optional[List[str]]
The list of required credentials for the provider.
For example, ["api_key", "api_secret"] which would be used inside
the fetcher to get the credentials from the credentials dict.
For example credentials.get("PROVIDER_API_KEY").
fetcher_list : List[Fetcher]
The list of fetchers that the provider supports.
"""
"""Initialize the provider."""
self.name = name
self.description = description
self.required_credentials = required_credentials
the_dict: Dict[str, Any] = {}
for query in self.QUERIES:
the_dict[query] = {getattr(x, query)().__name__: x for x in fetcher_list}
self.fetcher_dict = the_dict
# This is used only for documentation generation
self.formatted_credentials = []
if required_credentials is not None:
for rq in required_credentials:
self.formatted_credentials.append(f"{self.name.lower()}_{rq}")
self.fetcher_list = fetcher_list
self.fetcher_dict: Dict[str, Fetcher] = {}
self.fetcher_dict = {
str(fetcher.__name__).lower(): fetcher for fetcher in fetcher_list
}

349
openbb_sdk/sdk/provider/openbb_provider/map.py
View File

@@ -1,349 +0,0 @@
"""Create a detailed Map of the Provider Registry."""
from typing import Dict, Tuple
from openbb_provider.registry import provider_registry
def _get_data_type(fetcher) -> str:
"""Get the data type of the fetcher."""
data_type = (
str(fetcher.get_data_type()) # pylint: disable=use-maxsplit-arg
.split(".")[-1]
.replace("]", "")
.replace("'>", "")
)
return data_type
def _get_query_type(fetcher) -> str:
"""Get the query type of the fetcher."""
query_type = (
str(fetcher.get_query_type()) # pylint: disable=use-maxsplit-arg
.split(".")[-1]
.replace("'>", "")
)
return query_type
def _get_data_fields(fetcher) -> Tuple[Dict, Dict]:
"""Get the standard and provider-specific data fields of the fetcher.
Parameters
----------
fetcher : Fetcher
The fetcher to get the data fields from.
Returns
-------
Tuple[Dict, Dict]
A tuple of the provider-specific and standardized data fields.
"""
# TODO: this is a hacky way to get the fields, but it works for now
# if we split data into standard and extra, should be easier
try:
provider_fields = (fetcher.get_provider_data_type()).__fields__
except AttributeError:
try:
provider_fields = (fetcher.get_provider_data_type().__args__[0]).__fields__
except AttributeError:
provider_fields = (
(fetcher.get_data_type().__args__[-1]).__args__[0].__fields__
)
try:
standard_fields = (fetcher.get_data_type()).__fields__
except AttributeError:
try:
standard_fields = (fetcher.get_data_type().__args__[0]).__fields__
except AttributeError:
standard_fields = (
(fetcher.get_data_type().__args__[-1]).__args__[0].__fields__
)
return provider_fields, standard_fields
def _extract_fields(
provider_fields: Dict,
standard_fields: Dict,
) -> Tuple[Dict, Dict]:
"""Extract the fields and their details from pydantic models.
Parameters
----------
provider_fields : Dict
The fields of the provider pydantic model.
standard_fields : Dict
The fields of the standardized pydantic model.
Returns
-------
Tuple[Dict, Dict]
A tuple of the provider fields and standardized fields.
"""
provider_dict: Dict = {}
standard_dict: Dict = {}
for k, v in provider_fields.items():
if k in standard_fields:
standard_dict[k] = {
"type": v.annotation,
"default": v.default,
"required": v.required,
"description": v.field_info.description,
}
else:
provider_dict[k] = {
"type": v.annotation,
"default": v.default,
"required": v.required,
"description": v.field_info.description,
}
return provider_dict, standard_dict
def build_query_params_mapping() -> Dict:
"""Build a mapping of the provider registry QueryParams.
Returns
-------
Dict :
The mapping is a nested dictionary with the following structure:
{
"QueryParamsType": {
"provider_name": {
"param_name": {
"type": "param_type",
"default": "param_default",
"required": "param_required"
}
}
"openbb": {
"param_name": {
"type": "param_type",
"default": "param_default",
"required": "param_required"
}
}
}
}
"""
provider_registry_mapping = provider_registry.providers
mapping = {}
for provider in provider_registry_mapping.values():
for fetcher in provider.fetcher_list:
provider_name = provider.name
query_type = _get_query_type(fetcher)
provider_param_fields = (fetcher.get_provider_query_type()).__fields__
standard_param_fields = (fetcher.get_query_type()).__fields__
provider_specific_query_params, standardized_query_params = _extract_fields(
provider_param_fields,
standard_param_fields,
)
if query_type not in mapping:
mapping[query_type] = {provider_name: provider_specific_query_params}
mapping[query_type]["openbb"] = standardized_query_params # type: ignore
else:
mapping[query_type][provider_name] = provider_specific_query_params
return mapping
def build_data_mapping() -> Dict:
"""Build a mapping of the provider registry Data.
Returns
-------
Dict :
The mapping is a nested dictionary with the following structure:
{
"DataType": {
"provider_name": {
"data_field_name": {
"type": "param_type",
"default": "param_default",
"required": "param_required"
}
}
"openbb": {
"data_field_name": {
"type": "param_type",
"default": "param_default",
"required": "param_required"
}
}
}
}
"""
provider_registry_mapping = provider_registry.providers
mapping = {}
for provider in provider_registry_mapping.values():
for fetcher in provider.fetcher_list:
provider_name = provider.name
data_type = _get_data_type(fetcher)
provider_data_fields, standard_data_fields = _get_data_fields(fetcher)
provider_specific_data_fields, standardized_data_fields = _extract_fields(
provider_data_fields,
standard_data_fields,
)
if data_type not in mapping:
mapping[data_type] = {provider_name: provider_specific_data_fields}
mapping[data_type]["openbb"] = standardized_data_fields # type: ignore
else:
mapping[data_type][provider_name] = provider_specific_data_fields
return mapping
def _get_data_docstring(fetcher) -> Tuple[str, str]:
"""Get the data docstring of the fetcher for the provider and standardized data.
Parameters
----------
fetcher : Fetcher
The fetcher to get the data docstring from.
Returns
-------
Tuple[str, str]
A tuple of the provider and standardized data docstrings.
"""
provider_data_docstring = (fetcher.get_provider_data_type()).__doc__
if not provider_data_docstring:
try:
provider_data_docstring = (
fetcher.get_provider_data_type().__args__[0]
).__doc__
except AttributeError:
provider_data_docstring = ""
standard_data_docstring = (fetcher.get_data_type()).__doc__
if not standard_data_docstring:
try:
standard_data_docstring = (fetcher.get_data_type().__args__[0]).__doc__
except AttributeError:
standard_data_docstring = ""
return provider_data_docstring, standard_data_docstring
def build_docstring_mapping() -> Dict:
"""Build a mapping of the provider registry docstrings.
Returns
-------
Dict :
The mapping is a nested dictionary with the following structure:
{
"DataType": {
"provider_name": "docstring",
"openbb": "docstring"
},
"QueryParamsType": {
"provider_name": "docstring",
"openbb": "docstring"
}
}
"""
mapping = {}
provider_registry_mapping = provider_registry.providers
for provider in provider_registry_mapping.values():
for fetcher in provider.fetcher_list:
provider_name = provider.name
data_type = _get_data_type(fetcher)
query_type = _get_query_type(fetcher)
provider_query_docstring = (fetcher.get_provider_query_type()).__doc__
standard_query_docstring = (fetcher.get_query_type()).__doc__
provider_data_docstring, standard_data_docstring = _get_data_docstring(
fetcher
)
if data_type not in mapping:
mapping[data_type] = {
provider_name: provider_data_docstring,
"openbb": standard_data_docstring,
}
else:
mapping[data_type][provider_name] = provider_data_docstring
if query_type not in mapping:
mapping[query_type] = {
provider_name: provider_query_docstring,
"openbb": standard_query_docstring,
}
else:
mapping[query_type][provider_name] = provider_query_docstring
return mapping
def build_credentials_mapping() -> Dict:
"""Build a mapping of the provider registry credentials.
Returns
-------
Dict :
The mapping is a nested dictionary with the following structure:
{
"provider_name": [
"credential_name",
...
]
}
...
"""
mapping = {}
provider_registry_credentials = provider_registry.credentials
for provider, provider_credentials in provider_registry_credentials.items():
mapping[provider] = list(provider_credentials.keys())
return mapping
def build_model_mapping() -> Dict:
"""Build a mapping of the provider registry models.
Returns
-------
Dict :
The mapping is a nested dictionary of data and query mappings.
"""
mapping: Dict = {}
data_mapping = build_data_mapping()
query_mapping = build_query_params_mapping()
docstring_mapping = build_docstring_mapping()
for query_type, providers in query_mapping.items():
query_type_name = query_type.replace("QueryParams", "")
mapping[query_type_name] = {}
for provider, fields in providers.items():
mapping[query_type_name][provider] = {
"QueryParams": {},
"Data": {},
}
mapping[query_type_name][provider]["QueryParams"] = {
"fields": fields,
"docstring": docstring_mapping[query_type][provider],
}
for data_type, providers in data_mapping.items():
data_type_name = data_type.replace("Data", "")
for provider, fields in providers.items():
mapping[data_type_name][provider]["Data"] = {
"fields": fields,
"docstring": docstring_mapping[data_type][provider],
}
return mapping

64
openbb_sdk/sdk/provider/openbb_provider/query_executor.py Normal file
View File

@@ -0,0 +1,64 @@
from typing import Any, Dict, Optional
from openbb_provider.abstract.provider import Provider
from openbb_provider.registry import Registry, RegistryLoader
class ProviderError(Exception):
pass
class QueryExecutor:
"""Class to execute queries from providers"""
def __init__(self, registry: Optional[Registry] = None) -> None:
self.registry = registry or RegistryLoader.from_extensions()
def get_provider(self, provider_name: str):
"""Get a provider from the registry."""
# raise exception if provider not found
return self.registry.providers[provider_name.lower()]
def get_fetcher(self, provider: Provider, model_name: str):
"""Get a fetcher from a provider"""
# raise exception if fetcher not found
fetcher_name = (provider.name + model_name + "fetcher").lower()
return provider.fetcher_dict[fetcher_name]
def match_credentials(
self, provider: Provider, credentials: Optional[Dict[str, str]]
) -> Dict[str, str]:
"""Filter received credentials to match provider requirements"""
if provider.required_credentials is None:
return {}
if credentials is None:
credentials = {}
result: Dict[str, str] = {}
for c in provider.formatted_credentials:
credential_value = credentials.get(c)
if c not in credentials or credential_value is None:
raise ProviderError(f"Missing credential '{c}' for '{provider.name}'.")
result[c] = credential_value
return result
def execute(
self,
provider_name: str,
model_name: str,
params: Dict[str, Any],
credentials: Optional[Dict[str, str]] = None,
):
"""Execute query"""
provider = self.get_provider(provider_name)
Fetcher_ = self.get_fetcher(provider, model_name)
matched_credentials = self.match_credentials(provider, credentials)
try:
return Fetcher_.fetch_data(params, matched_credentials)
except Exception as e:
raise ProviderError(e) from e

163
openbb_sdk/sdk/provider/openbb_provider/registry.py
View File

@@ -1,141 +1,38 @@
"""ProviderRegistry class and Builder class."""
from typing import Any, Dict, Optional
from openbb_provider.abstract.provider import Provider
import pkg_resources
from openbb_provider.abstract.data import QueryParams
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.abstract.provider import Provider
class Registry:
"""Class to maintain registry of providers"""
def __init__(self) -> None:
self._providers = {}
@property
def providers(self):
return self._providers
def include_provider(self, name: str, provider: Provider) -> None:
self._providers[name.lower()] = provider
class ProviderError(Exception):
class ExtensionError(Exception):
pass
# TODO: This ProviderRegistry needs some refactoring. The two attributes partially
# repeat themselves.
class RegistryLoader:
@staticmethod
def from_extensions() -> Registry:
registry = Registry()
for entry_point in pkg_resources.iter_entry_points("openbb_provider_extension"):
try:
registry.include_provider(
name=entry_point.name,
provider=entry_point.load(),
)
except Exception as e:
raise ExtensionError(
f"Invalid provider '{entry_point.name}': {e}"
) from e
class ProviderRegistry:
"""A Provider Registry a class that holds all the providers and their credentials."""
def __init__(
self, providers: Dict[str, Provider], credentials: Dict[str, Dict[str, None]]
) -> None:
self.providers: Dict[str, Provider] = providers
self.credentials: Dict[str, Dict[str, None]] = credentials
def get_provider(self, provider_name: str) -> Provider:
"""Get a provider from the registry."""
if not self.providers:
raise ValueError("No providers found, please confirm they are loaded.")
try:
provider = self.providers[provider_name]
except KeyError as e:
raise ValueError(
f"{provider_name} is not part of the ProviderRegistry. "
f"Available providers are: {list(self.providers.keys())}"
) from e
return provider
def filter_credentials(
self,
provider_name: str,
credentials: Optional[Dict[str, str]] = None,
) -> Dict[str, str]:
required_credentials = self.credentials.get(provider_name)
if required_credentials is None:
return {}
result: Dict[str, str] = {}
if credentials is not None:
for k in required_credentials:
credential_value = credentials.get(k)
if k not in credentials or credential_value is None:
raise ProviderError(
f"Missing credential '{k}' for '{provider_name}'."
)
result[k] = credential_value
return result
def get_fetcher(self, provider: Provider, query_params: QueryParams) -> Fetcher:
"""Get a fetcher from the provider."""
fetch_dict = provider.fetcher_dict[
"get_query_type"
] # TODO: Check if we really need to pass this "get_query_type"
fetcher = fetch_dict.get(query_params.__name__) # type: ignore
if fetcher is None:
raise ProviderError(
f"This query is not supported by the '{provider.name}' provider. "
f"Please try another provider: '{list(self.providers.keys())}'"
)
return fetcher
def fetch(
self,
provider_name: str,
query_params: QueryParams,
extra_params: Optional[Dict] = None,
credentials: Optional[Dict[str, str]] = None,
):
"""Fetch data from a provider by using the OpenBB standard."""
provider = self.get_provider(provider_name)
filtered_credentials = self.filter_credentials(provider_name, credentials)
Fetcher_ = self.get_fetcher(provider, query_params)
try:
return Fetcher_.fetch_data(query_params, extra_params, filtered_credentials)
except Exception as e:
raise ProviderError(e) from e
def load_extensions(
entry_point_group: str = "openbb_provider_extension",
) -> Dict[str, Dict[str, Any]]:
"""Load extensions from entry points and their API keys from settings."""
# TODO: This function is overcomplicated and needs refactoring.
extensions_dict: Dict[str, Dict[str, Any]] = {}
extensions_dict["providers"] = {}
extensions_dict["credentials"] = {}
entry_points = pkg_resources.iter_entry_points(entry_point_group)
for entry_point in entry_points:
e = entry_point.load()
provider_name = str(e.name)
extensions_dict["providers"][provider_name] = e
required_credentials = getattr(e, "required_credentials", None)
if required_credentials:
extensions_dict["credentials"][provider_name] = {}
for credential in required_credentials:
if credential:
credential_name = provider_name.lower() + "_" + credential.lower()
extensions_dict["credentials"][provider_name][
credential_name
] = None
return extensions_dict
def build_provider_registry(
extensions_dict: Dict[str, Dict[str, Any]]
) -> ProviderRegistry:
"""Build a ProviderRegistry from a list of extensions and their API keys."""
return ProviderRegistry(
extensions_dict["providers"], extensions_dict["credentials"]
)
extensions_dict__ = load_extensions("openbb_provider_extension")
provider_registry = build_provider_registry(extensions_dict=extensions_dict__)
return registry

94
openbb_sdk/sdk/provider/openbb_provider/registry_map.py Normal file
View File

@@ -0,0 +1,94 @@
from typing import Dict, List, Literal, Optional
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.registry import Registry, RegistryLoader
class RegistryMap:
"""Class to store information about providers in a register."""
def __init__(self, registry: Optional[Registry] = None) -> None:
self._registry = registry or RegistryLoader.from_extensions()
self._required_credentials = self.get_required_credentials()
self._available_providers = self.get_available_providers()
self._map = self.get_map()
self._models = self.get_models(self._map)
@property
def registry(self) -> Registry:
return self._registry
@property
def available_providers(self) -> List[str]:
return self._available_providers
@property
def required_credentials(self) -> Dict[str, List[str]]:
return self._required_credentials
@property
def map(self) -> Dict[str, List[str]]:
return self._map
@property
def models(self) -> List[str]:
return self._models
def get_required_credentials(self) -> List[str]:
cred_list = []
for provider in self._registry.providers.values():
for c in provider.formatted_credentials:
cred_list.append(c)
return cred_list
def get_available_providers(self) -> List[str]:
return list(self._registry.providers.keys())
def get_map(self) -> Dict[str, List[str]]:
map_ = {}
for p in self._registry.providers:
for fetcher in self._registry.providers[p].fetcher_dict.values():
f = fetcher()
model_name = self.extract_model_name(f)
standard_query, extra_query = self.extract_fields(f, "query_params")
standard_data, extra_data = self.extract_fields(f, "data")
if model_name not in map_:
map_[model_name] = {}
map_[model_name]["openbb"] = {
"QueryParams": standard_query,
"Data": standard_data,
}
map_[model_name][p] = {
"QueryParams": extra_query,
"Data": extra_data,
}
return map_
def get_models(self, map_: Dict[str, List[str]]) -> List[str]:
return list(map_.keys())
@staticmethod
def extract_model_name(fetcher: Fetcher) -> str:
# TODO: We should have a better way to extract the model name.
# This looks fragile, but it's the best we can do for now.
return fetcher.query_params_type.__name__.replace("QueryParams", "")
@staticmethod
def extract_fields(fetcher: Fetcher, type_: Literal["query_params", "data"]):
standard_fields = getattr(fetcher, f"{type_}_type").__fields__
extra_fields = getattr(fetcher, f"provider_{type_}_type").__fields__
standard = {}
extra = {}
for name, field in extra_fields.items():
if name in standard_fields:
standard[name] = field
else:
extra[name] = field
return standard, extra