Source code for

from datetime import datetime
from os.path import exists, join
import json
import warnings

from .publisher import PublisherSet
from .dataset import DatasetSet
from .activity import ActivitySet
from .organisation import OrganisationSet
from ..utils.exceptions import NoDataError
from ..utils.config import CONFIG

[docs]class Registry(object): """Class representing the IATI registry.""" def __init__(self, path=None): """Construct a new Registry object. A warning is raised if the data is more than 7 days old. A ``NoDataError`` is raised if there is no data. """ self._last_updated = None self.path = path if path else CONFIG['paths']['registry'] last_updated = self.last_updated days_ago = ( - last_updated).days if days_ago > 7: warning_msg = 'Warning: Data was last updated {} days ' + \ 'ago. Consider downloading a fresh ' + \ 'data dump, using:\n\n ' + \ '>>>\n' warnings.warn(warning_msg.format(days_ago)) @property def last_updated(self): """Return the datetime when the local cache was last updated. """ if not self._last_updated: filepath = join(self.path, 'metadata.json') if exists(filepath): with open(filepath) as handler: j = json.load(handler) last_updated = j['updated_at'] self._last_updated = datetime.strptime( last_updated, '%Y-%m-%dT%H:%M:%SZ') else: error_msg = 'Error: No data found! ' + \ 'Download a fresh data dump ' + \ 'using:\n\n ' + \ '>>>\n' raise NoDataError(error_msg) return self._last_updated @property def publishers(self): """Return an iterator of all publishers on the registry.""" data_path = join(self.path, 'data', '*') metadata_path = join(self.path, 'metadata', '*') return PublisherSet(data_path, metadata_path) @property def datasets(self): """Return an iterator of all IATI datasets on the registry.""" publisher_set = self.publishers data_path = join(publisher_set.data_path, '*') metadata_path = join(publisher_set.metadata_path, '*') return DatasetSet(data_path, metadata_path) @property def activities(self): """Return an iterator of all IATI activities on the registry.""" return ActivitySet(self.datasets) @property def organisations(self): """Return an iterator of all IATI organisations on the registry.""" return OrganisationSet(self.datasets)