IdentityPython / pyFF / 513

Committed 24 Jan 2019 - 20:49 coverage: 82.502%. First build

Build # 513

Build Type

Pull #158

travis-ci

Committed by

web-flow

Commit Message

utf-8 border in store implementation

Pull Request Pull Request #158: Py3 compat

Run Details

135 of 157 new or added lines in 11 files covered. (85.99%)

2744 of 3326 relevant lines covered (82.5%)

2.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.88

/src/pyff/fetch.py

"""

An abstraction layer for metadata fetchers. Supports both syncronous and asyncronous fetchers with cache.

"""

from .logs import get_log
import os
import requests
from .constants import config
from datetime import datetime
from collections import deque
import six
from concurrent import futures
import traceback
from .parse import parse_resource
from itertools import chain
from .exceptions import ResourceException
from .utils import url_get

if six.PY2:
    from UserDict import DictMixin as ResourceManagerBase
elif six.PY3:
    from collections import MutableMapping as ResourceManagerBase


requests.packages.urllib3.disable_warnings()

log = get_log(__name__)


class ResourceManager(ResourceManagerBase):

    def __init__(self):
        self._resources = dict()
        self.shutdown = False

    def __setitem__(self, key, value):
        if not isinstance(value, Resource):
            raise ValueError("I can only store Resources")
        self._resources[key] = value

    def __getitem__(self, key):
        return self._resources[key]

    def __delitem__(self, key):
        if key in self:
            del self._resources[key]

    def keys(self):
        return list(self._resources.keys())

    def values(self):
        return list(self._resources.values())

    def walk(self, url=None):
        if url is not None:
            return self[url].walk()
        else:
            i = [r.walk() for r in list(self.values())]
            return chain(*i)

    def add(self, r):
        if not isinstance(r, Resource):
            raise ValueError("I can only store Resources")
        self[r.name] = r

    def __contains__(self, item):
        return item in self._resources

    def __len__(self):
        return len(list(self.values()))

    def __iter__(self):
        return self.walk()

    def reload(self, url=None, fail_on_error=False, store=None):
        # type: (object, basestring) -> None
        if url is not None:
            resources = deque([self[url]])
        else:
            resources = deque(list(self.values()))

        with futures.ThreadPoolExecutor(max_workers=config.worker_pool_size) as executor:
            while resources:
                tasks = dict((executor.submit(r.fetch, store=store), r) for r in resources)
                new_resources = deque()
                for future in futures.as_completed(tasks):
                    r = tasks[future]
                    try:
                        res = future.result()
                        if res is not None:
                            for nr in res:
                                new_resources.append(nr)
                    except Exception as ex:
                        log.debug(traceback.format_exc())
                        log.error(ex)
                        if fail_on_error:
                            raise ex
                resources = new_resources


class Resource(object):
    def __init__(self, url, **kwargs):
        self.url = url
        self.opts = kwargs
        self.t = None
        self.type = "text/plain"
        self.expire_time = None
        self.last_seen = None
        self._infos = deque(maxlen=config.info_buffer_size)
        self.children = deque()

        def _null(t):
            return t

        self.opts.setdefault('cleanup', _null)
        self.opts.setdefault('via', _null)
        self.opts.setdefault('fail_on_error', False)
        self.opts.setdefault('as', None)
        self.opts.setdefault('verify', None)
        self.opts.setdefault('filter_invalid', True)
        self.opts.setdefault('validate', True)

        if "://" not in self.url:
            if os.path.isfile(self.url):
                self.url = "file://{}".format(os.path.abspath(self.url))

    @property
    def post(self):
        return self.opts['via']

    @property
    def cleanup(self):
        return self.opts['cleanup']

    def __str__(self):
        return "Resource {} expires at {} using ".format(self.url, self.expire_time) + \
               ",".join(["{}={}".format(k, v) for k, v in list(self.opts.items())])

    def walk(self):
        yield self
        for c in self.children:
            for cn in c.walk():
                yield cn

    def is_expired(self):
        now = datetime.now()
        return self.expire_time is not None and self.expire_time < now

    def is_valid(self):
        return self.t is not None and not self.is_expired()

    def add_info(self, info):
        self._infos.append(info)

    def add_child(self, url, **kwargs):
        opts = dict()
        opts.update(self.opts)
        del opts['as']
        opts.update(kwargs)
        self.children.append(Resource(url, **opts))

    @property
    def name(self):
        if 'as' in self.opts:
            return self.opts['as']
        else:
            return self.url

    @property
    def info(self):
        if self._infos is None or not self._infos:
            return dict()
        else:
            return self._infos[-1]

    def fetch(self, store=None):
        info = dict()
        info['Resource'] = self.url
        self.add_info(info)
        data = None

        if os.path.isdir(self.url):
            data = self.url
            info['Directory'] = self.url
        elif '://' in self.url:
            r = url_get(self.url)

            info['HTTP Response Headers'] = r.headers
            log.debug("got status_code={:d}, encoding={} from_cache={} from {}".
                      format(r.status_code, r.encoding, getattr(r, "from_cache", False), self.url))
            info['Status Code'] = str(r.status_code)
            info['Reason'] = r.reason

            if r.ok:
                data = r.text
            else:
                raise ResourceException("Got status={:d} while fetching {}".format(r.status_code, self.url))
        else:
            raise ResourceException("Unknown resource type {}".format(self.url))

        parse_info = parse_resource(self, data)
        if parse_info is not None and isinstance(parse_info, dict):
            info.update(parse_info)

        if self.t is not None:
            self.last_seen = datetime.now()
            if self.post is not None:
                self.t = self.post(self.t, **self.opts)

            if self.is_expired():
                info['Expired'] = True
                raise ResourceException("Resource at {} expired on {}".format(self.url, self.expire_time))
            else:
                info['Expired'] = False

            for (eid, error) in list(info['Validation Errors'].items()):
                log.error(error)

            if store is not None:
                store.update(self.t, tid=self.name)

        return self.children

1	"""
2
3	An abstraction layer for metadata fetchers. Supports both syncronous and asyncronous fetchers with cache.
4
5	"""
6
7	from .logs import get_log	3×
8	import os	3×
9	import requests	3×
10	from .constants import config	3×
11	from datetime import datetime	3×
12	from collections import deque	3×
13	import six	3×
14	from concurrent import futures	3×
15	import traceback	3×
16	from .parse import parse_resource	3×
17	from itertools import chain	3×
18	from .exceptions import ResourceException	3×
19	from .utils import url_get	3×
20
21	if six.PY2:	3×
22	from UserDict import DictMixin as ResourceManagerBase	!
23	elif six.PY3:	3×
24	from collections import MutableMapping as ResourceManagerBase	3×
25
26
27	requests.packages.urllib3.disable_warnings()	3×
28
29	log = get_log(__name__)	3×
30
31
32	class ResourceManager(ResourceManagerBase):	3×
33
34	def __init__(self):	3×
35	self._resources = dict()	3×
36	self.shutdown = False	3×
37
38	def __setitem__(self, key, value):	3×
39	if not isinstance(value, Resource):	3×
40	raise ValueError("I can only store Resources")	!
41	self._resources[key] = value	3×
42
43	def __getitem__(self, key):	3×
44	return self._resources[key]	!
45
46	def __delitem__(self, key):	3×
47	if key in self:	!
48	del self._resources[key]	!
49
50	def keys(self):	3×
51	return list(self._resources.keys())	!
52
53	def values(self):	3×
54	return list(self._resources.values())	3×
55
56	def walk(self, url=None):	3×
57	if url is not None:	3×
58	return self[url].walk()	!
59	else:
60	i = [r.walk() for r in list(self.values())]	3×
61	return chain(*i)	3×
62
63	def add(self, r):	3×
64	if not isinstance(r, Resource):	3×
65	raise ValueError("I can only store Resources")	!
66	self[r.name] = r	3×
67
68	def __contains__(self, item):	3×
69	return item in self._resources	!
70
71	def __len__(self):	3×
NEW 72	return len(list(self.values()))	!
73
74	def __iter__(self):	3×
75	return self.walk()	!
76
77	def reload(self, url=None, fail_on_error=False, store=None):	3×
78	# type: (object, basestring) -> None
79	if url is not None:	3×
80	resources = deque([self[url]])	!
81	else:
82	resources = deque(list(self.values()))	3×
83
84	with futures.ThreadPoolExecutor(max_workers=config.worker_pool_size) as executor:	3×
85	while resources:	3×
86	tasks = dict((executor.submit(r.fetch, store=store), r) for r in resources)	3×
87	new_resources = deque()	3×
88	for future in futures.as_completed(tasks):	3×
89	r = tasks[future]	3×
90	try:	3×
91	res = future.result()	3×
92	if res is not None:	3×
93	for nr in res:	3×
94	new_resources.append(nr)	3×
95	except Exception as ex:	3×
96	log.debug(traceback.format_exc())	3×
97	log.error(ex)	3×
98	if fail_on_error:	3×
99	raise ex	3×
100	resources = new_resources	3×
101
102
103	class Resource(object):	3×
104	def __init__(self, url, **kwargs):	3×
105	self.url = url	3×
106	self.opts = kwargs	3×
107	self.t = None	3×
108	self.type = "text/plain"	3×
109	self.expire_time = None	3×
110	self.last_seen = None	3×
111	self._infos = deque(maxlen=config.info_buffer_size)	3×
112	self.children = deque()	3×
113
114	def _null(t):	3×
115	return t	!
116
117	self.opts.setdefault('cleanup', _null)	3×
118	self.opts.setdefault('via', _null)	3×
119	self.opts.setdefault('fail_on_error', False)	3×
120	self.opts.setdefault('as', None)	3×
121	self.opts.setdefault('verify', None)	3×
122	self.opts.setdefault('filter_invalid', True)	3×
123	self.opts.setdefault('validate', True)	3×
124
125	if "://" not in self.url:	3×
126	if os.path.isfile(self.url):	3×
127	self.url = "file://{}".format(os.path.abspath(self.url))	3×
128
129	@property	3×
130	def post(self):
131	return self.opts['via']	3×
132
133	@property	3×
134	def cleanup(self):
135	return self.opts['cleanup']	!
136
137	def __str__(self):	3×
138	return "Resource {} expires at {} using ".format(self.url, self.expire_time) + \	!
139	",".join(["{}={}".format(k, v) for k, v in list(self.opts.items())])
140
141	def walk(self):	3×
142	yield self	3×
143	for c in self.children:	3×
144	for cn in c.walk():	!
145	yield cn	!
146
147	def is_expired(self):	3×
148	now = datetime.now()	3×
149	return self.expire_time is not None and self.expire_time < now	3×
150
151	def is_valid(self):	3×
152	return self.t is not None and not self.is_expired()	!
153
154	def add_info(self, info):	3×
155	self._infos.append(info)	3×
156
157	def add_child(self, url, **kwargs):	3×
158	opts = dict()	3×
159	opts.update(self.opts)	3×
160	del opts['as']	3×
161	opts.update(kwargs)	3×
162	self.children.append(Resource(url, **opts))	3×
163
164	@property	3×
165	def name(self):
166	if 'as' in self.opts:	3×
167	return self.opts['as']	3×
168	else:
169	return self.url	!
170
171	@property	3×
172	def info(self):
173	if self._infos is None or not self._infos:	3×
174	return dict()	!
175	else:
176	return self._infos[-1]	3×
177
178	def fetch(self, store=None):	3×
179	info = dict()	3×
180	info['Resource'] = self.url	3×
181	self.add_info(info)	3×
182	data = None	3×
183
184	if os.path.isdir(self.url):	3×
185	data = self.url	3×
186	info['Directory'] = self.url	3×
187	elif '://' in self.url:	3×
188	r = url_get(self.url)	3×
189
190	info['HTTP Response Headers'] = r.headers	3×
191	log.debug("got status_code={:d}, encoding={} from_cache={} from {}".	3×
192	format(r.status_code, r.encoding, getattr(r, "from_cache", False), self.url))
193	info['Status Code'] = str(r.status_code)	3×
194	info['Reason'] = r.reason	3×
195
196	if r.ok:	3×
197	data = r.text	3×
198	else:
199	raise ResourceException("Got status={:d} while fetching {}".format(r.status_code, self.url))	3×
200	else:
201	raise ResourceException("Unknown resource type {}".format(self.url))	3×
202
203	parse_info = parse_resource(self, data)	3×
204	if parse_info is not None and isinstance(parse_info, dict):	3×
205	info.update(parse_info)	3×
206
207	if self.t is not None:	3×
208	self.last_seen = datetime.now()	3×
209	if self.post is not None:	3×
210	self.t = self.post(self.t, **self.opts)	3×
211
212	if self.is_expired():	3×
213	info['Expired'] = True	3×
214	raise ResourceException("Resource at {} expired on {}".format(self.url, self.expire_time))	3×
215	else:
216	info['Expired'] = False	3×
217
218	for (eid, error) in list(info['Validation Errors'].items()):	3×
219	log.error(error)	!
220
221	if store is not None:	3×
222	store.update(self.t, tid=self.name)	3×
223
224	return self.children	3×

IdentityPython / pyFF / 513

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous