from datetime import datetime
from typing import Any
from fews_openapi_py_client import AuthenticatedClient, Client
from fews_openapi_py_client.api.locations import locations
from fews_openapi_py_client.api.parameters import parameters
from fews_openapi_py_client.api.tasks import taskruns
from fews_openapi_py_client.api.timeseries import timeseries
from fews_openapi_py_client.api.whatif import post_what_if_scenarios
from fews_openapi_py_client.api.workflows import workflows
from fews_py_wrapper._api.base import ApiEndpoint
from fews_py_wrapper.utils import format_datetime
__all__ = [
"Taskruns",
"Parameters",
"Locations",
"TimeSeries",
"WhatIfScenarios",
"Workflows",
]
[docs]
class Taskruns(ApiEndpoint):
endpoint_function = staticmethod(taskruns.sync_detailed)
[docs]
def execute(
self,
*,
client: AuthenticatedClient | Client,
**kwargs: Any,
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
return super().execute(client=client, **kwargs)
[docs]
class Parameters(ApiEndpoint):
endpoint_function = staticmethod(parameters.sync_detailed)
[docs]
def execute(
self, *, client: AuthenticatedClient | Client, **kwargs: Any
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
return super().execute(client=client, **kwargs)
[docs]
class Locations(ApiEndpoint):
endpoint_function = staticmethod(locations.sync_detailed)
[docs]
def execute(
self, *, client: AuthenticatedClient | Client, **kwargs: Any
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
return super().execute(client=client, **kwargs)
[docs]
class TimeSeries(ApiEndpoint):
endpoint_function = staticmethod(timeseries.sync_detailed)
[docs]
def execute(
self, *, client: AuthenticatedClient | Client, **kwargs: Any
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
kwargs = self._format_time_args(kwargs)
return super().execute(client=client, **kwargs)
def _format_time_args(self, kwargs: dict[str, Any]) -> dict[str, Any]:
time_args = [
"start_time",
"end_time",
"start_creation_time",
"end_creation_time",
"start_forecast_time",
"end_forecast_time",
]
for arg in time_args:
if arg in kwargs and kwargs[arg] is not None:
if not isinstance(kwargs[arg], datetime):
arg_type = type(kwargs[arg])
raise ValueError(
f"Invalid argument value for {arg}: Expected datetime,"
f" got {arg_type}"
)
kwargs[arg] = format_datetime(kwargs[arg])
return kwargs
[docs]
class WhatIfScenarios(ApiEndpoint):
endpoint_function = staticmethod(post_what_if_scenarios.sync_detailed)
[docs]
def execute(
self, *, client: AuthenticatedClient | Client, **kwargs: Any
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
return super().execute(client=client, **kwargs)
[docs]
class Workflows(ApiEndpoint):
endpoint_function = staticmethod(workflows.sync_detailed)
[docs]
def execute(
self,
*,
client: AuthenticatedClient | Client,
**kwargs: Any,
) -> dict[str, Any]:
kwargs = self.update_input_kwargs(kwargs)
return super().execute(client=client, **kwargs)