Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions bcb/sgs/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,3 +137,44 @@ def get(codes, start=None, end=None, last=0, multi=True, freq=None):
return pd.concat(dfs, axis=1)
else:
return dfs


def get_json(code: int, start=None, end=None, last: int = 0) -> str:
"""
Retorna um JSON com séries temporais obtidas do SGS.

Parameters
----------

code : int
Código da série temporal
start : str, int, date, datetime, Timestamp
Data de início da série.
Interpreta diferentes tipos e formatos de datas.
end : string, int, date, datetime, Timestamp
Data final da série.
Interpreta diferentes tipos e formatos de datas.
last : int
Retorna os últimos ``last`` elementos disponíveis da série temporal
solicitada. Se ``last`` for maior que 0 (zero) os argumentos ``start``
e ``end`` são ignorados.

Returns
-------

JSON :
série temporal univariada em formato JSON.
"""
urd = _get_url_and_payload(code, start, end, last)
res = requests.get(urd["url"], params=urd["payload"])
if res.status_code != 200:
try:
res_json = json.loads(res.text)
except Exception:
res_json = {}
if "error" in res_json:
raise Exception("BCB error: {}".format(res_json["error"]))
elif "erro" in res_json:
raise Exception("BCB error: {}".format(res_json["erro"]["detail"]))
raise Exception("Download error: code = {}".format(code))
return res.text
32 changes: 32 additions & 0 deletions tests/sgs/test_series.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,3 +102,35 @@ def test_get_long_series_error():
)
else:
assert False, "Expected an exception but none was raised."


def test_json_return():
# Test for JSON return
x = sgs.get_json(1, last=10)
assert isinstance(x, str)
assert len(x) > 0
assert x.startswith("[")
assert x.endswith("]")


def test_json_return_long_series_error():
# Test for JSON return long series error
try:
sgs.get_json(1, start="2000-01-01", end="2023-01-01")
except Exception as e:
assert (
str(e)
== "BCB error: O sistema aceita uma janela de consulta de, no máximo, 10 anos em séries de periodicidade diária"
)
else:
assert False, "Expected an exception but none was raised."

try:
sgs.get_json(1, last=50)
except Exception as e:
assert (
str(e)
== "BCB error: br.gov.bcb.pec.sgs.comum.excecoes.SGSNegocioException: A quantidade máxima de valores deve ser 20"
)
else:
assert False, "Expected an exception but none was raised."