Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self.section in self.config:
data = self.config[self.section]
else:
data = self.config['DEFAULT']
keys = sorted(data.keys())
expected = sorted(self._configured_attribute_translations.values())
if keys != expected:
for key in expected:
if (key not in keys and
key not in self._non_default_configured_attributes):
raise _error.InvalidFeedConfig(
setting=key, feed=self,
message='missing configuration key: {}'.format(key))
for key in keys:
if key not in expected:
raise _error.InvalidFeedConfig(
setting=key, feed=self,
message='extra configuration key: {}'.format(key))
data = dict(
(self._configured_attribute_inverse_translations[k],
self._get_configured_attribute_value(
attribute=self._configured_attribute_inverse_translations[k],
key=k, data=data))
for k in data.keys())
for attr in self._non_default_configured_attributes:
if attr not in data:
data[attr] = None
self.__dict__.update(data)
def _fetch(self):
"""Fetch and parse a feed using feedparser.
>>> feed = Feed(
... name='test-feed',
... url='http://feeds.feedburner.com/allthingsrss/hJBr')
>>> parsed = feed._fetch()
>>> parsed.status
200
"""
_LOG.info('fetch {}'.format(self))
if not self.url:
raise _error.InvalidFeedConfig(setting='url', feed=self)
if self.section in self.config:
config = self.config[self.section]
else:
config = self.config['DEFAULT']
proxy = config['proxy']
timeout = config.getint('feed-timeout')
kwargs = {}
if proxy:
kwargs['handlers'] = [_urllib_request.ProxyHandler({'http':proxy})]
f = _util.TimeLimitedFunction(timeout, _feedparser.parse)
return f(self.url, self.etag, modified=self.modified, **kwargs)
message = 'error with feed {}'.format(feed.name)
super(FeedError, self).__init__(message=message, **kwargs)
self.feed = feed
class InvalidFeedConfig (FeedError):
def __init__(self, setting, feed, message=None, **kwargs):
if not message:
message = "invalid feed configuration {}".format(
{setting: getattr(feed, setting)})
super(InvalidFeedConfig, self).__init__(
feed=feed, message=message, **kwargs)
self.setting = setting
class InvalidFeedName (InvalidFeedConfig):
def __init__(self, name, message=None, **kwargs):
if not message:
message = 'invalid feed name {!r}'.format(name)
super(InvalidFeedName, self).__init__(
setting='name', message=message, **kwargs)
class DuplicateFeedName (InvalidFeedName):
def __init__(self, name, **kwargs):
message = 'duplicate feed name {!r}'.format(name)
super(DuplicateFeedName, self).__init__(
name=name, message=message, **kwargs)
class ProcessingError (FeedError):
def __init__(self, parsed, feed, message=None, **kwargs):
def __init__(self, setting, feed, message=None, **kwargs):
if not message:
message = "invalid feed configuration {}".format(
{setting: getattr(feed, setting)})
super(InvalidFeedConfig, self).__init__(
feed=feed, message=message, **kwargs)
self.setting = setting
def load_from_config(self, config=None):
"Restore configured attributes"
if config is None:
config = _config.CONFIG
self.config = config
if self.section in self.config:
data = self.config[self.section]
else:
data = self.config['DEFAULT']
keys = sorted(data.keys())
expected = sorted(self._configured_attribute_translations.values())
if keys != expected:
for key in expected:
if (key not in keys and
key not in self._non_default_configured_attributes):
raise _error.InvalidFeedConfig(
setting=key, feed=self,
message='missing configuration key: {}'.format(key))
for key in keys:
if key not in expected:
raise _error.InvalidFeedConfig(
setting=key, feed=self,
message='extra configuration key: {}'.format(key))
data = dict(
(self._configured_attribute_inverse_translations[k],
self._get_configured_attribute_value(
attribute=self._configured_attribute_inverse_translations[k],
key=k, data=data))
for k in data.keys())
for attr in self._non_default_configured_attributes:
if attr not in data:
data[attr] = None
super(DataFileError, self).__init__(feeds=feeds, message=message)
class NoDataFile (DataFileError):
def __init__(self, feeds):
message = 'feed data file {} does not exist'.format(feeds.datafile)
super(NoDataFile, self).__init__(feeds=feeds, message=message)
def log(self):
super(NoDataFile, self).log()
_LOG.warning(
"if you're using r2e for the first time, you have to run "
"'r2e new' first.")
class NoToEmailAddress (InvalidFeedConfig, FeedsError):
def __init__(self, feed, **kwargs):
message = 'no target email address has been defined'
super(NoToEmailAddress, self).__init__(
setting='to', feed=feed, message=message, **kwargs)
def log(self):
super(NoToEmailAddress, self).log()
_LOG.warning(
"please run 'r2e email emailaddress' or "
"'r2e add name url emailaddress'.")
class FeedIndexError (FeedsError, IndexError):
def __init__(self, index, message=None, **kwargs):
if message is None:
message = 'feed {!r} not found'.format(index)