"""BriteCore v2 Data API endpoint wrappers.
This module provides wrappers for data export and dashboard-discovery endpoints
in the BriteCore v2 data API.
"""
from typing import Any, Unpack
from britecore_sdk.api.api_calls import (
BritecoreAPIClient,
RequestParameters,
api_client,
)
from britecore_sdk.api.api_calls.v2._common import build_payload, post
API_CLIENT: BritecoreAPIClient = api_client
[docs]
def export_data_as_csv(
as_of_date: str | None = None,
end_date: str | None = None,
nonprep_dfs: str | None = None,
prep_dfs: str | None = None,
start_date: str | None = None,
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Export CSV data for the requested date range and data sets.
POST /api/v2/data/export_data_as_csv
"""
return post(
"/api/v2/data/export_data_as_csv",
build_payload(
as_of_date=as_of_date,
end_date=end_date,
nonprep_dfs=nonprep_dfs,
prep_dfs=prep_dfs,
start_date=start_date,
),
**kwargs,
)
[docs]
def get_available_dashboards(
module: str | None = None,
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Retrieve dashboards available for a module.
POST /api/v2/data/get_available_dashboards
"""
return post(
"/api/v2/data/get_available_dashboards",
build_payload(module=module),
**kwargs,
)
__all__ = [
"export_data_as_csv",
"get_available_dashboards",
]
# --- Autogenerated spec wrappers ---
[docs]
def fetch_json_report_result(
params: dict[str, Any] | None = None,
report_name: str | None = None,
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Fetch Json Report Result.
POST /api/v2/data/fetch_json_report_result
"""
request_json: dict[str, Any] = {
"params": params,
"report_name": report_name,
}
filtered_json = {k: v for k, v in request_json.items() if v is not None}
request_result = API_CLIENT.do_request(
path="/api/v2/data/fetch_json_report_result",
json=filtered_json,
method="POST",
**kwargs,
)
return API_CLIENT.process_result(
request_result, endpoint="/api/v2/data/fetch_json_report_result"
)
[docs]
def get_logical_catalog(
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Get Logical Catalog.
POST /api/v2/data/get_logical_catalog
"""
request_json: dict[str, Any] = {}
filtered_json = {k: v for k, v in request_json.items() if v is not None}
request_result = API_CLIENT.do_request(
path="/api/v2/data/get_logical_catalog",
json=filtered_json,
method="POST",
**kwargs,
)
return API_CLIENT.process_result(
request_result, endpoint="/api/v2/data/get_logical_catalog"
)
[docs]
def preview_sql_check(
query: str | None = None,
params: dict[str, Any] | None = None,
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Preview Sql Check.
POST /api/v2/data/preview_sql_check
"""
request_json: dict[str, Any] = {
"query": query,
"params": params,
}
filtered_json = {k: v for k, v in request_json.items() if v is not None}
request_result = API_CLIENT.do_request(
path="/api/v2/data/preview_sql_check",
json=filtered_json,
method="POST",
**kwargs,
)
return API_CLIENT.process_result(
request_result, endpoint="/api/v2/data/preview_sql_check"
)
[docs]
def preview_sql_result(
query: str | None = None,
params: dict[str, Any] | None = None,
**kwargs: Unpack[RequestParameters],
) -> Any:
"""Preview Sql Result.
POST /api/v2/data/preview_sql_result
"""
request_json: dict[str, Any] = {
"query": query,
"params": params,
}
filtered_json = {k: v for k, v in request_json.items() if v is not None}
request_result = API_CLIENT.do_request(
path="/api/v2/data/preview_sql_result",
json=filtered_json,
method="POST",
**kwargs,
)
return API_CLIENT.process_result(
request_result, endpoint="/api/v2/data/preview_sql_result"
)
__all__.extend(
[
"fetch_json_report_result",
"get_logical_catalog",
"preview_sql_check",
"preview_sql_result",
]
)