Doapi metadata enrichment

.gitignore

@@ -22,6 +22,7 @@ _build
 # Distribution / packaging
 .Python
 env/
+.venv/
 build/
 develop-eggs/
 dist/

.travis.yml

@@ -9,4 +9,4 @@ install:
 script:
 - tox
 after_success:
-- coveralls
+- coveralls

cartoframes/__init__.py

@@ -5,7 +5,7 @@
 
 
 # Check installed packages versions
-check_package('carto', '>=1.8.3')
+check_package('carto', '>=1.9.1')
 check_package('pandas', '>=0.23.0')
 check_package('geopandas', '>=0.6.0')
 

cartoframes/data/observatory/catalog/dataset.py

@@ -4,7 +4,7 @@
 from shapely import wkt
 
 from .entity import CatalogEntity
-from .repository.dataset_repo import get_dataset_repo
+from .repository.dataset_repo import get_dataset_repo, DATASET_TYPE
 from .repository.geography_repo import get_geography_repo
 from .repository.variable_repo import get_variable_repo
 from .repository.variable_group_repo import get_variable_group_repo
@@ -17,8 +17,6 @@
 from ....utils.utils import get_credentials, check_credentials, check_do_enabled
 from ....exceptions import DOError
 
-DATASET_TYPE = 'dataset'
-
 
 class Dataset(CatalogEntity):
     """A Dataset represents the metadata of a particular dataset in the catalog.
@@ -375,7 +373,7 @@ def _join_geographies_geodataframes(geographies_gdf1, geographies_gdf2):
         return join_gdf['id'].unique()
 
     @check_do_enabled
-    def to_csv(self, file_path, credentials=None, limit=None):
+    def to_csv(self, file_path, credentials=None, limit=None, order_by=None):
         """Download dataset data as a local csv file. You need Data Observatory enabled in your CARTO
         account, please contact us at support@carto.com for more information.
 
@@ -402,10 +400,10 @@ def to_csv(self, file_path, credentials=None, limit=None):
             raise DOError('You are not subscribed to this Dataset yet. '
                           'Please, use the subscribe method first.')
 
-        self._download(_credentials, file_path, limit)
+        self._download(_credentials, file_path, limit, order_by)
 
     @check_do_enabled
-    def to_dataframe(self, credentials=None, limit=None):
+    def to_dataframe(self, credentials=None, limit=None, order_by=None):
         """Download dataset data as a pandas.DataFrame. You need Data Observatory enabled in your CARTO
         account, please contact us at support@carto.com for more information.
 
@@ -434,7 +432,7 @@ def to_dataframe(self, credentials=None, limit=None):
             raise DOError('You are not subscribed to this Dataset yet. '
                           'Please, use the subscribe method first.')
 
-        return self._download(_credentials, limit=limit)
+        return self._download(_credentials, limit=limit, order_by=order_by)
 
     @check_do_enabled
     def subscribe(self, credentials=None):
@@ -468,7 +466,7 @@ def subscribe(self, credentials=None):
 
         """
         _credentials = get_credentials(credentials)
-        _subscribed_ids = subscriptions.get_subscription_ids(_credentials)
+        _subscribed_ids = subscriptions.get_subscription_ids(_credentials, DATASET_TYPE)
 
         if self.id in _subscribed_ids:
             utils.display_existing_subscription_message(self.id, DATASET_TYPE)

cartoframes/data/observatory/catalog/entity.py

@@ -109,7 +109,7 @@ def _get_print_id(self):
 
         return self.id
 
-    def _download(self, credentials, file_path=None, limit=None):
+    def _download(self, credentials, file_path=None, limit=None, order_by=None):
         if not self._is_available_in('bq'):
             raise DOError('{} is not ready for Download. Please, contact us for more information.'.format(self))
 
@@ -126,6 +126,8 @@ def _download(self, credentials, file_path=None, limit=None):
         column_names = bq_client.get_table_column_names(project, dataset, table)
 
         query = 'SELECT * FROM `{}`'.format(full_remote_table_name)
+        if order_by:
+            query = '{} ORDER BY {}'.format(query, order_by)
         if limit:
             query = '{} LIMIT {}'.format(query, limit)
 

cartoframes/data/observatory/catalog/geography.py

@@ -1,15 +1,13 @@
 from .entity import CatalogEntity
 from .repository.dataset_repo import get_dataset_repo
-from .repository.geography_repo import get_geography_repo
+from .repository.geography_repo import get_geography_repo, GEOGRAPHY_TYPE
 from .repository.constants import GEOGRAPHY_FILTER
 from . import subscription_info
 from . import subscriptions
 from . import utils
 from ....utils.utils import get_credentials, check_credentials, check_do_enabled
 from ....exceptions import DOError
 
-GEOGRAPHY_TYPE = 'geography'
-
 
 class Geography(CatalogEntity):
     """A Geography represents the metadata of a particular geography dataset in the catalog.
@@ -178,7 +176,7 @@ def get_all(cls, filters=None, credentials=None):
         return cls._entity_repo.get_all(filters, credentials)
 
     @check_do_enabled
-    def to_csv(self, file_path, credentials=None, limit=None):
+    def to_csv(self, file_path, credentials=None, limit=None, order_by=None):
         """Download geography data as a local csv file. You need Data Observatory enabled in your CARTO
         account, please contact us at support@carto.com for more information.
 
@@ -205,10 +203,10 @@ def to_csv(self, file_path, credentials=None, limit=None):
             raise DOError('You are not subscribed to this Geography yet. '
                           'Please, use the subscribe method first.')
 
-        self._download(_credentials, file_path, limit)
+        self._download(_credentials, file_path, limit, order_by)
 
     @check_do_enabled
-    def to_dataframe(self, credentials=None, limit=None):
+    def to_dataframe(self, credentials=None, limit=None, order_by=None):
         """Download geography data as a pandas.DataFrame. You need Data Observatory enabled in your CARTO
         account, please contact us at support@carto.com for more information.
 
@@ -237,7 +235,7 @@ def to_dataframe(self, credentials=None, limit=None):
             raise DOError('You are not subscribed to this Geography yet. '
                           'Please, use the subscribe method first.')
 
-        return self._download(_credentials, limit=limit)
+        return self._download(_credentials, limit=limit, order_by=order_by)
 
     @check_do_enabled
     def subscribe(self, credentials=None):
@@ -270,7 +268,7 @@ def subscribe(self, credentials=None):
 
         """
         _credentials = get_credentials(credentials)
-        _subscribed_ids = subscriptions.get_subscription_ids(_credentials)
+        _subscribed_ids = subscriptions.get_subscription_ids(_credentials, GEOGRAPHY_TYPE)
 
         if self.id in _subscribed_ids:
             utils.display_existing_subscription_message(self.id, GEOGRAPHY_TYPE)

cartoframes/data/observatory/catalog/repository/category_repo.py

@@ -21,9 +21,6 @@ def _get_entity_class(cls):
         return Category
 
     def _get_rows(self, filters=None):
-        if filters is not None and COUNTRY_FILTER in filters.keys():
-            return self.client.get_categories_joined_datasets(filters)
-
         return self.client.get_categories(filters)
 
     def _map_row(self, row):

cartoframes/data/observatory/catalog/repository/constants.py

@@ -1,7 +1,7 @@
-CATEGORY_FILTER = 'category_id'
-COUNTRY_FILTER = 'country_id'
-DATASET_FILTER = 'dataset_id'
-GEOGRAPHY_FILTER = 'geography_id'
-PROVIDER_FILTER = 'provider_id'
-VARIABLE_FILTER = 'variable_id'
-VARIABLE_GROUP_FILTER = 'variable_group_id'
+CATEGORY_FILTER = 'category'
+COUNTRY_FILTER = 'country'
+DATASET_FILTER = 'dataset'
+GEOGRAPHY_FILTER = 'geography'
+PROVIDER_FILTER = 'provider'
+VARIABLE_FILTER = 'variable'
+VARIABLE_GROUP_FILTER = 'variable_group'

cartoframes/data/observatory/catalog/repository/country_repo.py

@@ -2,7 +2,7 @@
 from .entity_repo import EntityRepository
 
 
-_COUNTRY_ID_FIELD = 'country_id'
+_COUNTRY_ID_FIELD = 'id'
 _ALLOWED_FILTERS = [CATEGORY_FILTER]
 
 

cartoframes/data/observatory/catalog/repository/dataset_repo.py

@@ -1,11 +1,11 @@
-from .constants import CATEGORY_FILTER, COUNTRY_FILTER, GEOGRAPHY_FILTER, PROVIDER_FILTER, VARIABLE_FILTER
+from .constants import CATEGORY_FILTER, COUNTRY_FILTER, GEOGRAPHY_FILTER, PROVIDER_FILTER
 from .entity_repo import EntityRepository
-from ..entity import CatalogList
 
+DATASET_TYPE = 'dataset'
 
 _DATASET_ID_FIELD = 'id'
 _DATASET_SLUG_FIELD = 'slug'
-_ALLOWED_FILTERS = [CATEGORY_FILTER, COUNTRY_FILTER, GEOGRAPHY_FILTER, PROVIDER_FILTER, VARIABLE_FILTER]
+_ALLOWED_FILTERS = [CATEGORY_FILTER, COUNTRY_FILTER, GEOGRAPHY_FILTER, PROVIDER_FILTER]
 
 
 def get_dataset_repo():
@@ -18,10 +18,16 @@ def __init__(self):
         super(DatasetRepository, self).__init__(_DATASET_ID_FIELD, _ALLOWED_FILTERS, _DATASET_SLUG_FIELD)
 
     def get_all(self, filters=None, credentials=None):
+        if credentials is not None:
+            filters = self._add_subscription_ids(filters, credentials, DATASET_TYPE)
+            if filters is None:
+                return []
+
+        # Using user credentials to fetch entities
         self.client.set_user_credentials(credentials)
-        response = self._get_filtered_entities(filters)
-        self.client.set_user_credentials(None)
-        return response
+        entities = self._get_filtered_entities(filters)
+        self.client.reset_user_credentials()
+        return entities
 
     @classmethod
     def _get_entity_class(cls):
@@ -56,10 +62,5 @@ def _map_row(self, row):
             'available_in': self._normalize_field(row, 'available_in')
         }
 
-    def get_datasets_for_geographies(self, geographies):
-        rows = self.client.get_datasets_for_geographies(geographies)
-        normalized_data = [self._get_entity_class()(self._map_row(row)) for row in rows]
-        return CatalogList(normalized_data)
-
 
 _REPO = DatasetRepository()

cartoframes/data/observatory/catalog/repository/entity_repo.py

@@ -3,6 +3,7 @@
 from .repo_client import RepoClient
 from ..entity import CatalogList, is_slug_value
 from .....exceptions import CatalogError
+from ..subscriptions import get_subscription_ids
 
 
 def check_catalog_connection(method):
@@ -65,9 +66,9 @@ def _get_filters(self, filters):
 
     def _get_id_filter(self, id_):
         if self.slug_field is not None and is_slug_value(id_):
-            return {self.slug_field: id_}
+            return {self.slug_field: [id_]}
 
-        return {self.id_field: id_}
+        return {self.id_field: [id_]}
 
     def _get_id_list_filters(self, id_list):
         if self.slug_field is None:
@@ -91,16 +92,23 @@ def _get_id_list_filters(self, id_list):
 
         return filters
 
+    def _add_subscription_ids(self, filters, credentials, entity_type):
+        ids = get_subscription_ids(credentials, entity_type)
+
+        if not isinstance(ids, list) or len(ids) == 0:
+            return None
+
+        filters = filters or {}
+        filters['id'] = ids
+        return filters
+
     @classmethod
     def _to_catalog_entity(cls, result):
         return cls._get_entity_class()(result)
 
     @classmethod
     def _normalize_field(cls, row, field):
-        if field in row:
-            return row[field]
-
-        return None
+        return row.get(field, None)
 
     @classmethod
     @abstractmethod

cartoframes/data/observatory/catalog/repository/geography_repo.py

@@ -1,17 +1,15 @@
-from cartoframes.auth import Credentials
+from geopandas import GeoDataFrame
 
+from .....utils.geom_utils import set_geometry
 from .constants import COUNTRY_FILTER, CATEGORY_FILTER, PROVIDER_FILTER
 from .entity_repo import EntityRepository
 
-from .....io.carto import read_carto
-
+GEOGRAPHY_TYPE = 'geography'
 
 _GEOGRAPHY_ID_FIELD = 'id'
 _GEOGRAPHY_SLUG_FIELD = 'slug'
 _ALLOWED_FILTERS = [COUNTRY_FILTER, CATEGORY_FILTER, PROVIDER_FILTER]
 
-_DO_CREDENTIALS = Credentials('do-metadata', 'default_public')
-
 
 def get_geography_repo():
     return _REPO
@@ -23,20 +21,23 @@ def __init__(self):
         super(GeographyRepository, self).__init__(_GEOGRAPHY_ID_FIELD, _ALLOWED_FILTERS, _GEOGRAPHY_SLUG_FIELD)
 
     def get_all(self, filters=None, credentials=None):
+        if credentials is not None:
+            filters = self._add_subscription_ids(filters, credentials, GEOGRAPHY_TYPE)
+            if filters is None:
+                return []
+
+        # Using user credentials to fetch entities
         self.client.set_user_credentials(credentials)
-        response = self._get_filtered_entities(filters)
-        self.client.set_user_credentials(None)
-        return response
+        entities = self._get_filtered_entities(filters)
+        self.client.reset_user_credentials()
+        return entities
 
     @classmethod
     def _get_entity_class(cls):
         from cartoframes.data.observatory.catalog.geography import Geography
         return Geography
 
     def _get_rows(self, filters=None):
-        if filters is not None and (COUNTRY_FILTER in filters.keys() or CATEGORY_FILTER in filters.keys()):
-            return self.client.get_geographies_joined_datasets(filters)
-
         return self.client.get_geographies(filters)
 
     def _map_row(self, row):
@@ -59,8 +60,10 @@ def _map_row(self, row):
         }
 
     def get_geographies_gdf(self):
-        query = 'select id, geom_coverage as the_geom from geographies_public where geom_coverage is not null'
-        return read_carto(query, _DO_CREDENTIALS)
+        data = self.client.get_geographies({'get_geoms_coverage': True})
+        gdf = GeoDataFrame(data, crs='epsg:4326')
+        set_geometry(gdf, col='geom_coverage', inplace=True)
+        return gdf
 
 
 _REPO = GeographyRepository()