better splitting bind signature to its own class

This commit is contained in:
Diogo Sousa
2023-07-05 17:31:44 +01:00
parent 6eff39f4ad
commit e763fc745b
3 changed files with 60 additions and 57 deletions

View File

@@ -1,12 +1,6 @@
from typing import List, Literal
import numpy as np
from openbb_sdk_core.app.utils import (
basemodel_to_df,
df_to_basemodel,
get_target_column,
get_target_columns,
)
import pandas as pd
import pandas_ta as ta
import statsmodels.api as sm
@@ -25,6 +19,12 @@ from openbb_provider.model.abstract.data import Data
from openbb_sdk_core.app.model.command_output import CommandOutput
from openbb_sdk_core.app.model.results.empty import Empty
from openbb_sdk_core.app.router import Router
from openbb_sdk_core.app.utils import (
basemodel_to_df,
df_to_basemodel,
get_target_column,
get_target_columns,
)
from pydantic import NonNegativeFloat, PositiveInt
from scipy import stats
from statsmodels.tsa import stattools
@@ -356,7 +356,7 @@ def so(
results = ((returns - target_return) / downside_deviation).dropna()
if adjusted:
results = result / np.sqrt(2)
results = results / np.sqrt(2)
results = df_to_basemodel(results)

View File

@@ -1,10 +1,4 @@
from typing import List, Literal, Optional
from openbb_sdk_core.app.utils import (
basemodel_to_df,
df_to_basemodel,
get_target_column,
get_target_columns,
)
import pandas as pd
from builtin_extensions.common.ta import ta_helpers
@@ -13,6 +7,12 @@ from openbb_sdk_core.app.model.command_output import CommandOutput
from openbb_sdk_core.app.model.export.plotly import Plotly
from openbb_sdk_core.app.model.results.empty import Empty
from openbb_sdk_core.app.router import Router
from openbb_sdk_core.app.utils import (
basemodel_to_df,
df_to_basemodel,
get_target_column,
get_target_columns,
)
from openbb_sdk_core.plots.plots import YTimeSeries, plot_timeseries
from pydantic import NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt

View File

@@ -163,14 +163,14 @@ class Router:
kwargs["response_model_exclude_unset"] = True
kwargs["openapi_extra"] = {"query": query}
func = self.bind_signature(func, query)
func = SignatureTransformer.complete_signature(func, query)
CommandValidator.check(func=func)
kwargs["path"] = kwargs.get("path", f"/{func.__name__}")
kwargs["endpoint"] = func
kwargs["methods"] = kwargs.get("methods", ["GET"])
kwargs["description"] = self.slice_docstring(func)
kwargs["description"] = SignatureTransformer.slice_docstring(func)
api_router.add_api_route(**kwargs)
@@ -186,6 +186,50 @@ class Router:
router=router.api_router, prefix=prefix, tags=tags # type: ignore
)
class SignatureTransformer:
@classmethod
def complete_signature(
cls, func: Callable[P, CommandOutput], query: str
) -> Callable[P, CommandOutput]:
"""Complete function signature."""
provider_interface = get_provider_interface()
if query:
if query not in provider_interface.queries:
raise AttributeError(
f"Invalid query: {query}. Check available queries in ProviderInterface().queries"
)
cls.validate_signature(func=func)
func = cls.inject_dependency(
func=func,
arg="provider_choices",
callable_=provider_interface.query_providers[query],
)
func = cls.inject_dependency(
func=func,
arg="standard_params",
callable_=provider_interface.params[query]["standard"],
)
func = cls.inject_dependency(
func=func,
arg="extra_params",
callable_=provider_interface.params[query]["extra"],
)
data_type = provider_interface.merged_data[query]
func.__annotations__["return"] = CommandOutput[List[data_type]] # type: ignore
elif (
"provider_choices" in func.__annotations__
and func.__annotations__["provider_choices"] == ProviderChoices
):
func.__annotations__["provider_choices"] = provider_interface.providers
return func
@staticmethod
def validate_signature(func: Callable[P, CommandOutput]) -> None:
"""Validate function signature before binding to query."""
@@ -228,50 +272,9 @@ class Router:
func.__annotations__[arg] = Annotated[callable_, Depends()] # type: ignore
return func
@staticmethod
def bind_signature(
func: Callable[P, CommandOutput], query: str
) -> Callable[P, CommandOutput]:
"""Bind function signature to a query."""
provider_interface = get_provider_interface()
if query:
if query not in provider_interface.queries:
raise AttributeError(
f"Invalid query: {query}. Check available queries in ProviderInterface().queries"
)
Router.validate_signature(func=func)
func = Router.inject_dependency(
func=func,
arg="provider_choices",
callable_=provider_interface.query_providers[query],
)
func = Router.inject_dependency(
func=func,
arg="standard_params",
callable_=provider_interface.params[query]["standard"],
)
func = Router.inject_dependency(
func=func,
arg="extra_params",
callable_=provider_interface.params[query]["extra"],
)
data_type = provider_interface.merged_data[query]
func.__annotations__["return"] = CommandOutput[List[data_type]] # type: ignore
elif (
"provider_choices" in func.__annotations__
and func.__annotations__["provider_choices"] == ProviderChoices
):
func.__annotations__["provider_choices"] = provider_interface.providers
return func
@staticmethod
def slice_docstring(func: Callable) -> str:
"""Slice docstring to remove Parameters and Returns sections."""
doc = func.__doc__
if doc:
sliced = doc.split(" Parameters\n ----------")[0]