Skip to content
Open
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
1 change: 1 addition & 0 deletions ipeaData/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from ipeaData.ipeadata import get_data, get_region_level, get_metadata, get_sources
112 changes: 68 additions & 44 deletions ipeaData/ipeadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,70 +3,94 @@
import pandas as pd


def basic_api_call(api):
def basic_api_call(url):
"""
Function to make a call on ipedata api
:param api: url on api
:return: a dataFrame with data
Calls IpeaData API.

:param url: (str) URL

:return: DataFrame with requested data
:rtype: pandas DataFrame
"""
response = req.get(api)
if response.status_code == req.codes.ok: # pylint: disable=no-member
response = req.get(url)

if response.status_code == req.codes.ok: # pylint: disable=no-member

json_response = response.json()
if 'value' in json_response:
try:
data_frame = pd.DataFrame(json_response['value'])
return data_frame
except Exception: # pylint: disable=broad-except
return None

df = pd.DataFrame(json_response['value'])

# Raises an error if the DataFrame is empty
if df.empty:
raise FileNotFoundError('We were unable to find data according to the input parameters.')

return df

return None


def get_sources():
"""
Get sources from ipea web site
:return: a data frame with the information
Lists sources from IpeaData containing the following fields:

1. FNTID: <FILL EXPLANATION>
2. FNTSIGLA: <FILL EXPLANATION>
3. MACRO: <FILL EXPLANATION>
4. REGIONAL: <FILL EXPLANATION>
5. SOCIAL: <FILL EXPLANATION>

:return: DataFrame with sources attributes
:rtype: pandas DataFrame
"""
api = "http://ipeadata2-homologa.ipea.gov.br/api/v1/Fontes"
return basic_api_call(api)
return basic_api_call("http://ipeadata2-homologa.ipea.gov.br/api/v1/Fontes")

def get_metadata(serie=None):

def get_metadata(id=None):
"""
Return metadata of a serie
:param serie: serie to search for otherwise return metadata for all series
:return: a data frame
Returns metadata for the given time series.

If no `id` parameter is passed, it returns metadata for all the time series in IpeaData.

:param id: (str) time series id

:return: Metadata for the given time series
:rtype: pandas DataFrame
"""
url_final = "('%s')" % serie if serie is not None else ''
api = "http://ipeadata2-homologa.ipea.gov.br/api/v1/Metadados%s" % url_final
return basic_api_call(api)
id = "('%s')" % id if id is not None else ''
return basic_api_call("http://ipeadata2-homologa.ipea.gov.br/api/v1/Metadados%s" % id)


def get_nivel_region(serie):
def get_region_level(id):
"""
Return region nivel of a serie
:param serie: serie to search for
:return: a data frame
Returns region level for the given time series.

:param id: (str) time series id

:return: Region level of a time series
:rtype: pandas DataFrame
"""
api = ("http://ipeadata2-homologa.ipea.gov.br/api/v1/Metadados('{}')"
"/Valores?$apply=groupby((NIVNOME))&$orderby=NIVNOME").format(serie)
return basic_api_call(api)
url = ("http://ipeadata2-homologa.ipea.gov.br/api/v1/Metadados('{}')/Valores?$apply=groupby((NIVNOME))&$orderby=NIV"
"NOME").format(id)
return basic_api_call(url)


# pylint: disable=invalid-name
def ipeadata(serie, groupby=None):
def get_data(id, groupby=None):
"""
Return the values from a given serie
:param serie: a serie to search for
:return: a data frame with the values
Returns data corresponding to the given time series.

:param id: (str) time series id

:return: Data for the given time series
:rtype: pandas DataFrame
"""
if groupby is not None:
df = get_nivel_region(serie)
df = get_region_level(id)
if df['NIVNOME'].isin([groupby]).any():
api = ("http://ipeadata2-homologa.ipea.gov.br/api/v1/AnoValors"
"(SERCODIGO='{}',NIVNOME='{}')?$top=100&$skip=0&$orderby"
"=SERATUALIZACAO&$count=true").format(serie, groupby)
return basic_api_call(api)
url = ("http://ipeadata2-homologa.ipea.gov.br/api/v1/AnoValors(SERCODIGO='{}',NIVNOME='{}')?$top=100&$skip="
"0&$orderby=SERATUALIZACAO&$count=true").format(id, groupby)
return basic_api_call(url)
return None
api = "http://ipeadata2-homologa.ipea.gov.br/api/v1/ValoresSerie(SERCODIGO='%s')" % serie
return basic_api_call(api)


if __name__ == "__main__":
print(ipeadata('ADMIS'))
url = "http://ipeadata2-homologa.ipea.gov.br/api/v1/ValoresSerie(SERCODIGO='%s')" % id
return basic_api_call(url)
12 changes: 7 additions & 5 deletions test/test_api.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,17 @@
from ipeaData.ipeadata import *
import unittest


class TestApi(unittest.TestCase):

def test_ipeadata(self):
self.assertIsNotNone(ipeadata('ADMIS'))
self.assertIsNotNone(get_data('ADMIS'))

def test_ipeadata_groupby(self):
df = ipeadata('COMP', 'Estados')
df = get_data('COMP', 'Estados')
self.assertIsNotNone(df)
self.assertEqual(27, df.shape[0])
df = ipeadata('COMP', 'nadanan')
df = get_data('COMP', 'nadanan')
self.assertIsNone(df)

def test_get_sources(self):
Expand All @@ -21,9 +22,10 @@ def test_get_metadata(self):
self.assertIsNotNone(get_metadata())

def test_get_region(self):
df = get_nivel_region('QUANTLEITE')
df = get_region_level('QUANTLEITE')
self.assertIsNotNone(df)
self.assertEqual(13, df.shape[0])


if __name__ == '__main__':
unittest.main(verbosity=3)
unittest.main(verbosity=3)