RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use the itemlaoders library (#4516)

This commit is contained in:
Júlio César Batista 2020-07-15 07:08:36 -03:00 committed by GitHub
parent eb1bc74417
commit 38496a00b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 854 additions and 1226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
docs/conf.py
View File

@ -284,6 +284,7 @@ intersphinx_mapping = {
'attrs': ('https://www.attrs.org/en/stable/', None),
'coverage': ('https://coverage.readthedocs.io/en/stable', None),
'cssselect': ('https://cssselect.readthedocs.io/en/latest', None),
'itemloaders': ('https://itemloaders.readthedocs.io/en/latest/', None),
'pytest': ('https://docs.pytest.org/en/latest', None),
'python': ('https://docs.python.org/3', None),
'sphinx': ('https://www.sphinx-doc.org/en/master', None),
@ -305,3 +306,15 @@ hoverxref_role_types = {
"ref": "tooltip",
}
hoverxref_roles = ['command', 'reqmeta', 'setting', 'signal']
def setup(app):
app.connect('autodoc-skip-member', maybe_skip_member)
def maybe_skip_member(app, what, name, obj, skip, options):
if not skip:
# autodocs was generating a text "alias of" for the following members
# https://github.com/sphinx-doc/sphinx/issues/4422
return name in {'default_item_class', 'default_selector_class'}
return skip

418
docs/topics/loaders.rst
View File

@ -20,6 +20,10 @@ Item Loaders are designed to provide a flexible, efficient and easy mechanism
for extending and overriding different field parsing rules, either by spider,
or by source format (HTML, XML, etc) without becoming a nightmare to maintain.
.. note:: Item Loaders are an extension of the itemloaders_ library that make it
easier to work with Scrapy by adding support for
:ref:`responses <topics-request-response>`.
Using Item Loaders to populate items
====================================
@ -173,8 +177,8 @@ The other thing you need to keep in mind is that the values returned by input
processors are collected internally (in lists) and then passed to output
processors to populate the fields.
Last, but not least, Scrapy comes with some :ref:`commonly used processors
<topics-loaders-available-processors>` built-in for convenience.
Last, but not least, itemloaders_ comes with some :ref:`commonly used
processors <itemloaders:built-in-processors>` built-in for convenience.
Declaring Item Loaders
@ -182,8 +186,8 @@ Declaring Item Loaders
Item Loaders are declared using a class definition syntax. Here is an example::
from itemloaders.processors import TakeFirst, MapCompose, Join
from scrapy.loader import ItemLoader
from scrapy.loader.processors import TakeFirst, MapCompose, Join
class ProductLoader(ItemLoader):
@ -214,7 +218,7 @@ output processors to use: in the :ref:`Item Field <topics-items-fields>`
metadata. Here is an example::
import scrapy
from scrapy.loader.processors import Join, MapCompose, TakeFirst
from itemloaders.processors import Join, MapCompose, TakeFirst
from w3lib.html import remove_tags
def filter_price(value):
@ -295,250 +299,9 @@ There are several ways to modify Item Loader context values:
ItemLoader objects
==================
.. class:: ItemLoader([item, selector, response], **kwargs)
Return a new Item Loader for populating the given :ref:`item object
<topics-items>`. If no item object is given, one is instantiated
automatically using the class in :attr:`default_item_class`.
When instantiated with a ``selector`` or a ``response`` parameters
the :class:`ItemLoader` class provides convenient mechanisms for extracting
data from web pages using :ref:`selectors <topics-selectors>`.
:param item: The item instance to populate using subsequent calls to
:meth:`~ItemLoader.add_xpath`, :meth:`~ItemLoader.add_css`,
or :meth:`~ItemLoader.add_value`.
:type item: :ref:`item object <topics-items>`
:param selector: The selector to extract data from, when using the
:meth:`add_xpath` (resp. :meth:`add_css`) or :meth:`replace_xpath`
(resp. :meth:`replace_css`) method.
:type selector: :class:`~scrapy.selector.Selector` object
:param response: The response used to construct the selector using the
:attr:`default_selector_class`, unless the selector argument is given,
in which case this argument is ignored.
:type response: :class:`~scrapy.http.Response` object
The item, selector, response and the remaining keyword arguments are
assigned to the Loader context (accessible through the :attr:`context` attribute).
:class:`ItemLoader` instances have the following methods:
.. method:: get_value(value, *processors, **kwargs)
Process the given ``value`` by the given ``processors`` and keyword
arguments.
Available keyword arguments:
:param re: a regular expression to use for extracting data from the
given value using :meth:`~scrapy.utils.misc.extract_regex` method,
applied before processors
:type re: str or compiled regex
Examples:
>>> from scrapy.loader.processors import TakeFirst
>>> loader.get_value(u'name: foo', TakeFirst(), unicode.upper, re='name: (.+)')
'FOO`
.. method:: add_value(field_name, value, *processors, **kwargs)
Process and then add the given ``value`` for the given field.
The value is first passed through :meth:`get_value` by giving the
``processors`` and ``kwargs``, and then passed through the
:ref:`field input processor <topics-loaders-processors>` and its result
appended to the data collected for that field. If the field already
contains collected data, the new data is added.
The given ``field_name`` can be ``None``, in which case values for
multiple fields may be added. And the processed value should be a dict
with field_name mapped to values.
Examples::
loader.add_value('name', u'Color TV')
loader.add_value('colours', [u'white', u'blue'])
loader.add_value('length', u'100')
loader.add_value('name', u'name: foo', TakeFirst(), re='name: (.+)')
loader.add_value(None, {'name': u'foo', 'sex': u'male'})
.. method:: replace_value(field_name, value, *processors, **kwargs)
Similar to :meth:`add_value` but replaces the collected data with the
new value instead of adding it.
.. method:: get_xpath(xpath, *processors, **kwargs)
Similar to :meth:`ItemLoader.get_value` but receives an XPath instead of a
value, which is used to extract a list of unicode strings from the
selector associated with this :class:`ItemLoader`.
:param xpath: the XPath to extract data from
:type xpath: str
:param re: a regular expression to use for extracting data from the
selected XPath region
:type re: str or compiled regex
Examples::
# HTML snippet: <p class="product-name">Color TV</p>
loader.get_xpath('//p[@class="product-name"]')
# HTML snippet: <p id="price">the price is $1200</p>
loader.get_xpath('//p[@id="price"]', TakeFirst(), re='the price is (.*)')
.. method:: add_xpath(field_name, xpath, *processors, **kwargs)
Similar to :meth:`ItemLoader.add_value` but receives an XPath instead of a
value, which is used to extract a list of unicode strings from the
selector associated with this :class:`ItemLoader`.
See :meth:`get_xpath` for ``kwargs``.
:param xpath: the XPath to extract data from
:type xpath: str
Examples::
# HTML snippet: <p class="product-name">Color TV</p>
loader.add_xpath('name', '//p[@class="product-name"]')
# HTML snippet: <p id="price">the price is $1200</p>
loader.add_xpath('price', '//p[@id="price"]', re='the price is (.*)')
.. method:: replace_xpath(field_name, xpath, *processors, **kwargs)
Similar to :meth:`add_xpath` but replaces collected data instead of
adding it.
.. method:: get_css(css, *processors, **kwargs)
Similar to :meth:`ItemLoader.get_value` but receives a CSS selector
instead of a value, which is used to extract a list of unicode strings
from the selector associated with this :class:`ItemLoader`.
:param css: the CSS selector to extract data from
:type css: str
:param re: a regular expression to use for extracting data from the
selected CSS region
:type re: str or compiled regex
Examples::
# HTML snippet: <p class="product-name">Color TV</p>
loader.get_css('p.product-name')
# HTML snippet: <p id="price">the price is $1200</p>
loader.get_css('p#price', TakeFirst(), re='the price is (.*)')
.. method:: add_css(field_name, css, *processors, **kwargs)
Similar to :meth:`ItemLoader.add_value` but receives a CSS selector
instead of a value, which is used to extract a list of unicode strings
from the selector associated with this :class:`ItemLoader`.
See :meth:`get_css` for ``kwargs``.
:param css: the CSS selector to extract data from
:type css: str
Examples::
# HTML snippet: <p class="product-name">Color TV</p>
loader.add_css('name', 'p.product-name')
# HTML snippet: <p id="price">the price is $1200</p>
loader.add_css('price', 'p#price', re='the price is (.*)')
.. method:: replace_css(field_name, css, *processors, **kwargs)
Similar to :meth:`add_css` but replaces collected data instead of
adding it.
.. method:: load_item()
Populate the item with the data collected so far, and return it. The
data collected is first passed through the :ref:`output processors
<topics-loaders-processors>` to get the final value to assign to each
item field.
.. method:: nested_xpath(xpath)
Create a nested loader with an xpath selector.
The supplied selector is applied relative to selector associated
with this :class:`ItemLoader`. The nested loader shares the :ref:`item
object <topics-items>` with the parent :class:`ItemLoader` so calls to
:meth:`add_xpath`, :meth:`add_value`, :meth:`replace_value`, etc. will
behave as expected.
.. method:: nested_css(css)
Create a nested loader with a css selector.
The supplied selector is applied relative to selector associated
with this :class:`ItemLoader`. The nested loader shares the :ref:`item
object <topics-items>` with the parent :class:`ItemLoader` so calls to
:meth:`add_xpath`, :meth:`add_value`, :meth:`replace_value`, etc. will
behave as expected.
.. method:: get_collected_values(field_name)
Return the collected values for the given field.
.. method:: get_output_value(field_name)
Return the collected values parsed using the output processor, for the
given field. This method doesn't populate or modify the item at all.
.. method:: get_input_processor(field_name)
Return the input processor for the given field.
.. method:: get_output_processor(field_name)
Return the output processor for the given field.
:class:`ItemLoader` instances have the following attributes:
.. attribute:: item
The :ref:`item object <topics-items>` being parsed by this Item Loader.
This is mostly used as a property so when attempting to override this
value, you may want to check out :attr:`default_item_class` first.
.. attribute:: context
The currently active :ref:`Context <topics-loaders-context>` of this
Item Loader.
.. attribute:: default_item_class
An :ref:`item object <topics-items>` class or factory, used to
instantiate items when not given in the ``__init__`` method.
.. attribute:: default_input_processor
The default input processor to use for those fields which don't specify
one.
.. attribute:: default_output_processor
The default output processor to use for those fields which don't specify
one.
.. attribute:: default_selector_class
The class used to construct the :attr:`selector` of this
:class:`ItemLoader`, if only a response is given in the ``__init__`` method.
If a selector is given in the ``__init__`` method this attribute is ignored.
This attribute is sometimes overridden in subclasses.
.. attribute:: selector
The :class:`~scrapy.selector.Selector` object to extract data from.
It's either the selector given in the ``__init__`` method or one created from
the response given in the ``__init__`` method using the
:attr:`default_selector_class`. This attribute is meant to be
read-only.
.. autoclass:: scrapy.loader.ItemLoader
:members:
:inherited-members:
.. _topics-loaders-nested:
@ -609,7 +372,7 @@ those dashes in the final product names.
Here's how you can remove those dashes by reusing and extending the default
Product Item Loader (``ProductLoader``)::
from scrapy.loader.processors import MapCompose
from itemloaders.processors import MapCompose
from myproject.ItemLoaders import ProductLoader
def strip_dashes(x):
@ -622,7 +385,7 @@ Another case where extending Item Loaders can be very helpful is when you have
multiple source formats, for example XML and HTML. In the XML version you may
want to remove ``CDATA`` occurrences. Here's an example of how to do it::
from scrapy.loader.processors import MapCompose
from itemloaders.processors import MapCompose
from myproject.ItemLoaders import ProductLoader
from myproject.utils.xml import remove_cdata
@ -642,156 +405,5 @@ projects. Scrapy only provides the mechanism; it doesn't impose any specific
organization of your Loaders collection - that's up to you and your project's
needs.
.. _topics-loaders-available-processors:
Available built-in processors
=============================
.. module:: scrapy.loader.processors
:synopsis: A collection of processors to use with Item Loaders
Even though you can use any callable function as input and output processors,
Scrapy provides some commonly used processors, which are described below. Some
of them, like the :class:`MapCompose` (which is typically used as input
processor) compose the output of several functions executed in order, to
produce the final parsed value.
Here is a list of all built-in processors:
.. class:: Identity
The simplest processor, which doesn't do anything. It returns the original
values unchanged. It doesn't receive any ``__init__`` method arguments, nor does it
accept Loader contexts.
Example:
>>> from scrapy.loader.processors import Identity
>>> proc = Identity()
>>> proc(['one', 'two', 'three'])
['one', 'two', 'three']
.. class:: TakeFirst
Returns the first non-null/non-empty value from the values received,
so it's typically used as an output processor to single-valued fields.
It doesn't receive any ``__init__`` method arguments, nor does it accept Loader contexts.
Example:
>>> from scrapy.loader.processors import TakeFirst
>>> proc = TakeFirst()
>>> proc(['', 'one', 'two', 'three'])
'one'
.. class:: Join(separator=u' ')
Returns the values joined with the separator given in the ``__init__`` method, which
defaults to ``u' '``. It doesn't accept Loader contexts.
When using the default separator, this processor is equivalent to the
function: ``u' '.join``
Examples:
>>> from scrapy.loader.processors import Join
>>> proc = Join()
>>> proc(['one', 'two', 'three'])
'one two three'
>>> proc = Join('<br>')
>>> proc(['one', 'two', 'three'])
'one<br>two<br>three'
.. class:: Compose(*functions, **default_loader_context)
A processor which is constructed from the composition of the given
functions. This means that each input value of this processor is passed to
the first function, and the result of that function is passed to the second
function, and so on, until the last function returns the output value of
this processor.
By default, stop process on ``None`` value. This behaviour can be changed by
passing keyword argument ``stop_on_none=False``.
Example:
>>> from scrapy.loader.processors import Compose
>>> proc = Compose(lambda v: v[0], str.upper)
>>> proc(['hello', 'world'])
'HELLO'
Each function can optionally receive a ``loader_context`` parameter. For
those which do, this processor will pass the currently active :ref:`Loader
context <topics-loaders-context>` through that parameter.
The keyword arguments passed in the ``__init__`` method are used as the default
Loader context values passed to each function call. However, the final
Loader context values passed to functions are overridden with the currently
active Loader context accessible through the :meth:`ItemLoader.context`
attribute.
.. class:: MapCompose(*functions, **default_loader_context)
A processor which is constructed from the composition of the given
functions, similar to the :class:`Compose` processor. The difference with
this processor is the way internal results are passed among functions,
which is as follows:
The input value of this processor is *iterated* and the first function is
applied to each element. The results of these function calls (one for each element)
are concatenated to construct a new iterable, which is then used to apply the
second function, and so on, until the last function is applied to each
value of the list of values collected so far. The output values of the last
function are concatenated together to produce the output of this processor.
Each particular function can return a value or a list of values, which is
flattened with the list of values returned by the same function applied to
the other input values. The functions can also return ``None`` in which
case the output of that function is ignored for further processing over the
chain.
This processor provides a convenient way to compose functions that only
work with single values (instead of iterables). For this reason the
:class:`MapCompose` processor is typically used as input processor, since
data is often extracted using the
:meth:`~scrapy.selector.Selector.extract` method of :ref:`selectors
<topics-selectors>`, which returns a list of unicode strings.
The example below should clarify how it works:
>>> def filter_world(x):
... return None if x == 'world' else x
...
>>> from scrapy.loader.processors import MapCompose
>>> proc = MapCompose(filter_world, str.upper)
>>> proc(['hello', 'world', 'this', 'is', 'scrapy'])
['HELLO, 'THIS', 'IS', 'SCRAPY']
As with the Compose processor, functions can receive Loader contexts, and
``__init__`` method keyword arguments are used as default context values. See
:class:`Compose` processor for more info.
.. class:: SelectJmes(json_path)
Queries the value using the json path provided to the ``__init__`` method and returns the output.
Requires jmespath (https://github.com/jmespath/jmespath.py) to run.
This processor takes only one input at a time.
Example:
>>> from scrapy.loader.processors import SelectJmes, Compose, MapCompose
>>> proc = SelectJmes("foo") #for direct use on lists and dictionaries
>>> proc({'foo': 'bar'})
'bar'
>>> proc({'foo': {'bar': 'baz'}})
{'bar': 'baz'}
Working with Json:
>>> import json
>>> proc_single_json_str = Compose(json.loads, SelectJmes("foo"))
>>> proc_single_json_str('{"foo": "bar"}')
'bar'
>>> proc_json_list = Compose(json.loads, MapCompose(SelectJmes('foo')))
>>> proc_json_list('[{"foo":"bar"}, {"baz":"tar"}]')
['bar']
.. _itemloaders: https://itemloaders.readthedocs.io/en/latest/
.. _processors: https://itemloaders.readthedocs.io/en/latest/built-in-processors.html

269
scrapy/loader/__init__.py
View File

@ -3,217 +3,86 @@ Item Loader
See documentation in docs/topics/loaders.rst
"""
from collections import defaultdict
from contextlib import suppress
from itemadapter import ItemAdapter
import itemloaders
from scrapy.item import Item
from scrapy.loader.common import wrap_loader_context
from scrapy.loader.processors import Identity
from scrapy.selector import Selector
from scrapy.utils.misc import arg_to_iter, extract_regex
from scrapy.utils.python import flatten
def unbound_method(method):
class ItemLoader(itemloaders.ItemLoader):
"""
Allow to use single-argument functions as input or output processors
(no need to define an unused first 'self' argument)
A user-friendly abstraction to populate an :ref:`item <topics-items>` with data
by applying :ref:`field processors <topics-loaders-processors>` to scraped data.
When instantiated with a ``selector`` or a ``response`` it supports
data extraction from web pages using :ref:`selectors <topics-selectors>`.
:param item: The item instance to populate using subsequent calls to
:meth:`~ItemLoader.add_xpath`, :meth:`~ItemLoader.add_css`,
or :meth:`~ItemLoader.add_value`.
:type item: scrapy.item.Item
:param selector: The selector to extract data from, when using the
:meth:`add_xpath`, :meth:`add_css`, :meth:`replace_xpath`, or
:meth:`replace_css` method.
:type selector: :class:`~scrapy.selector.Selector` object
:param response: The response used to construct the selector using the
:attr:`default_selector_class`, unless the selector argument is given,
in which case this argument is ignored.
:type response: :class:`~scrapy.http.Response` object
If no item is given, one is instantiated automatically using the class in
:attr:`default_item_class`.
The item, selector, response and remaining keyword arguments are
assigned to the Loader context (accessible through the :attr:`context` attribute).
.. attribute:: item
The item object being parsed by this Item Loader.
This is mostly used as a property so, when attempting to override this
value, you may want to check out :attr:`default_item_class` first.
.. attribute:: context
The currently active :ref:`Context <loaders-context>` of this Item Loader.
.. attribute:: default_item_class
An :ref:`item <topics-items>` class (or factory), used to instantiate
items when not given in the ``__init__`` method.
.. attribute:: default_input_processor
The default input processor to use for those fields which don't specify
one.
.. attribute:: default_output_processor
The default output processor to use for those fields which don't specify
one.
.. attribute:: default_selector_class
The class used to construct the :attr:`selector` of this
:class:`ItemLoader`, if only a response is given in the ``__init__`` method.
If a selector is given in the ``__init__`` method this attribute is ignored.
This attribute is sometimes overridden in subclasses.
.. attribute:: selector
The :class:`~scrapy.selector.Selector` object to extract data from.
It's either the selector given in the ``__init__`` method or one created from
the response given in the ``__init__`` method using the
:attr:`default_selector_class`. This attribute is meant to be
read-only.
"""
with suppress(AttributeError):
if '.' not in method.__qualname__:
return method.__func__
return method
class ItemLoader:
default_item_class = Item
default_input_processor = Identity()
default_output_processor = Identity()
default_selector_class = Selector
def __init__(self, item=None, selector=None, response=None, parent=None, **context):
if selector is None and response is not None:
selector = self.default_selector_class(response)
self.selector = selector
context.update(selector=selector, response=response)
if item is None:
item = self.default_item_class()
self.context = context
self.parent = parent
self._local_item = context['item'] = item
self._local_values = defaultdict(list)
# values from initial item
for field_name, value in ItemAdapter(item).items():
self._values[field_name] += arg_to_iter(value)
@property
def _values(self):
if self.parent is not None:
return self.parent._values
else:
return self._local_values
@property
def item(self):
if self.parent is not None:
return self.parent.item
else:
return self._local_item
def nested_xpath(self, xpath, **context):
selector = self.selector.xpath(xpath)
context.update(selector=selector)
subloader = self.__class__(
item=self.item, parent=self, **context
)
return subloader
def nested_css(self, css, **context):
selector = self.selector.css(css)
context.update(selector=selector)
subloader = self.__class__(
item=self.item, parent=self, **context
)
return subloader
def add_value(self, field_name, value, *processors, **kw):
value = self.get_value(value, *processors, **kw)
if value is None:
return
if not field_name:
for k, v in value.items():
self._add_value(k, v)
else:
self._add_value(field_name, value)
def replace_value(self, field_name, value, *processors, **kw):
value = self.get_value(value, *processors, **kw)
if value is None:
return
if not field_name:
for k, v in value.items():
self._replace_value(k, v)
else:
self._replace_value(field_name, value)
def _add_value(self, field_name, value):
value = arg_to_iter(value)
processed_value = self._process_input_value(field_name, value)
if processed_value:
self._values[field_name] += arg_to_iter(processed_value)
def _replace_value(self, field_name, value):
self._values.pop(field_name, None)
self._add_value(field_name, value)
def get_value(self, value, *processors, **kw):
regex = kw.get('re', None)
if regex:
value = arg_to_iter(value)
value = flatten(extract_regex(regex, x) for x in value)
for proc in processors:
if value is None:
break
_proc = proc
proc = wrap_loader_context(proc, self.context)
try:
value = proc(value)
except Exception as e:
raise ValueError("Error with processor %s value=%r error='%s: %s'" %
(_proc.__class__.__name__, value,
type(e).__name__, str(e)))
return value
def load_item(self):
adapter = ItemAdapter(self.item)
for field_name in tuple(self._values):
value = self.get_output_value(field_name)
if value is not None:
adapter[field_name] = value
return adapter.item
def get_output_value(self, field_name):
proc = self.get_output_processor(field_name)
proc = wrap_loader_context(proc, self.context)
try:
return proc(self._values[field_name])
except Exception as e:
raise ValueError("Error with output processor: field=%r value=%r error='%s: %s'" %
(field_name, self._values[field_name], type(e).__name__, str(e)))
def get_collected_values(self, field_name):
return self._values[field_name]
def get_input_processor(self, field_name):
proc = getattr(self, '%s_in' % field_name, None)
if not proc:
proc = self._get_item_field_attr(field_name, 'input_processor',
self.default_input_processor)
return unbound_method(proc)
def get_output_processor(self, field_name):
proc = getattr(self, '%s_out' % field_name, None)
if not proc:
proc = self._get_item_field_attr(field_name, 'output_processor',
self.default_output_processor)
return unbound_method(proc)
def _process_input_value(self, field_name, value):
proc = self.get_input_processor(field_name)
_proc = proc
proc = wrap_loader_context(proc, self.context)
try:
return proc(value)
except Exception as e:
raise ValueError(
"Error with input processor %s: field=%r value=%r "
"error='%s: %s'" % (_proc.__class__.__name__, field_name,
value, type(e).__name__, str(e)))
def _get_item_field_attr(self, field_name, key, default=None):
field_meta = ItemAdapter(self.item).get_field_meta(field_name)
return field_meta.get(key, default)
def _check_selector_method(self):
if self.selector is None:
raise RuntimeError("To use XPath or CSS selectors, "
"%s must be instantiated with a selector "
"or a response" % self.__class__.__name__)
def add_xpath(self, field_name, xpath, *processors, **kw):
values = self._get_xpathvalues(xpath, **kw)
self.add_value(field_name, values, *processors, **kw)
def replace_xpath(self, field_name, xpath, *processors, **kw):
values = self._get_xpathvalues(xpath, **kw)
self.replace_value(field_name, values, *processors, **kw)
def get_xpath(self, xpath, *processors, **kw):
values = self._get_xpathvalues(xpath, **kw)
return self.get_value(values, *processors, **kw)
def _get_xpathvalues(self, xpaths, **kw):
self._check_selector_method()
xpaths = arg_to_iter(xpaths)
return flatten(self.selector.xpath(xpath).getall() for xpath in xpaths)
def add_css(self, field_name, css, *processors, **kw):
values = self._get_cssvalues(css, **kw)
self.add_value(field_name, values, *processors, **kw)
def replace_css(self, field_name, css, *processors, **kw):
values = self._get_cssvalues(css, **kw)
self.replace_value(field_name, values, *processors, **kw)
def get_css(self, css, *processors, **kw):
values = self._get_cssvalues(css, **kw)
return self.get_value(values, *processors, **kw)
def _get_cssvalues(self, csss, **kw):
self._check_selector_method()
csss = arg_to_iter(csss)
return flatten(self.selector.css(css).getall() for css in csss)
context.update(response=response)
super().__init__(item=item, selector=selector, parent=parent, **context)

19
scrapy/loader/common.py
View File

@ -1,14 +1,21 @@
"""Common functions used in Item Loaders code"""
from functools import partial
from scrapy.utils.python import get_func_args
import warnings
from itemloaders import common
from scrapy.utils.deprecate import ScrapyDeprecationWarning
def wrap_loader_context(function, context):
"""Wrap functions that receive loader_context to contain the context
"pre-loaded" and expose a interface that receives only one argument
"""
if 'loader_context' in get_func_args(function):
return partial(function, loader_context=context)
else:
return function
warnings.warn(
"scrapy.loader.common.wrap_loader_context has moved to a new library."
"Please update your reference to itemloaders.common.wrap_loader_context",
ScrapyDeprecationWarning,
stacklevel=2
)
return common.wrap_loader_context(function, context)

99
scrapy/loader/processors.py
View File

@ -3,102 +3,19 @@ This module provides some commonly used processors for Item Loaders.
See documentation in docs/topics/loaders.rst
"""
from collections import ChainMap
from itemloaders import processors
from scrapy.utils.misc import arg_to_iter
from scrapy.loader.common import wrap_loader_context
from scrapy.utils.deprecate import create_deprecated_class
class MapCompose:
MapCompose = create_deprecated_class('MapCompose', processors.MapCompose)
def __init__(self, *functions, **default_loader_context):
self.functions = functions
self.default_loader_context = default_loader_context
Compose = create_deprecated_class('Compose', processors.Compose)
def __call__(self, value, loader_context=None):
values = arg_to_iter(value)
if loader_context:
context = ChainMap(loader_context, self.default_loader_context)
else:
context = self.default_loader_context
wrapped_funcs = [wrap_loader_context(f, context) for f in self.functions]
for func in wrapped_funcs:
next_values = []
for v in values:
try:
next_values += arg_to_iter(func(v))
except Exception as e:
raise ValueError("Error in MapCompose with "
"%s value=%r error='%s: %s'" %
(str(func), value, type(e).__name__,
str(e)))
values = next_values
return values
TakeFirst = create_deprecated_class('TakeFirst', processors.TakeFirst)
Identity = create_deprecated_class('Identity', processors.Identity)
class Compose:
SelectJmes = create_deprecated_class('SelectJmes', processors.SelectJmes)
def __init__(self, *functions, **default_loader_context):
self.functions = functions
self.stop_on_none = default_loader_context.get('stop_on_none', True)
self.default_loader_context = default_loader_context
def __call__(self, value, loader_context=None):
if loader_context:
context = ChainMap(loader_context, self.default_loader_context)
else:
context = self.default_loader_context
wrapped_funcs = [wrap_loader_context(f, context) for f in self.functions]
for func in wrapped_funcs:
if value is None and self.stop_on_none:
break
try:
value = func(value)
except Exception as e:
raise ValueError("Error in Compose with "
"%s value=%r error='%s: %s'" %
(str(func), value, type(e).__name__, str(e)))
return value
class TakeFirst:
def __call__(self, values):
for value in values:
if value is not None and value != '':
return value
class Identity:
def __call__(self, values):
return values
class SelectJmes:
"""
Query the input string for the jmespath (given at instantiation),
and return the answer
Requires : jmespath(https://github.com/jmespath/jmespath)
Note: SelectJmes accepts only one input element at a time.
"""
def __init__(self, json_path):
self.json_path = json_path
import jmespath
self.compiled_path = jmespath.compile(self.json_path)
def __call__(self, value):
"""Query value for the jmespath query and return answer
:param value: a data structure (dict, list) to extract from
:return: Element extracted according to jmespath query
"""
return self.compiled_path.search(value)
class Join:
def __init__(self, separator=u' '):
self.separator = separator
def __call__(self, values):
return self.separator.join(values)
Join = create_deprecated_class('Join', processors.Join)

6
scrapy/utils/misc.py
View File

@ -15,6 +15,7 @@ from w3lib.html import replace_entities
from scrapy.utils.datatypes import LocalWeakReferencedCache
from scrapy.utils.python import flatten, to_unicode
from scrapy.item import _BaseItem
from scrapy.utils.deprecate import ScrapyDeprecationWarning
_ITERABLE_SINGLE_VALUES = dict, _BaseItem, str, bytes
@ -86,6 +87,11 @@ def extract_regex(regex, text, encoding='utf-8'):
* if the regex contains multiple numbered groups, all those will be returned (flattened)
* if the regex doesn't contain any group the entire regex matching is returned
"""
warnings.warn(
"scrapy.utils.misc.extract_regex has moved to parsel.utils.extract_regex.",
ScrapyDeprecationWarning,
stacklevel=2
)
if isinstance(regex, str):
regex = re.compile(regex, re.UNICODE)

3
setup.cfg
View File

@ -118,6 +118,9 @@ ignore_errors = True
[mypy-tests.test_loader]
ignore_errors = True
[mypy-tests.test_loader_deprecated]
ignore_errors = True
[mypy-tests.test_pipeline_crawl]
ignore_errors = True

1
setup.py
View File

@ -71,6 +71,7 @@ setup(
'Twisted>=17.9.0',
'cryptography>=2.0',
'cssselect>=0.9.1',
'itemloaders>=1.0.1',
'lxml>=3.5.0',
'parsel>=1.5.0',
'PyDispatcher>=2.0.5',

1
tests/requirements-py3.txt
View File

@ -1,7 +1,6 @@
# Tests requirements
attrs
dataclasses; python_version == '3.6'
jmespath
mitmproxy; python_version >= '3.6'
mitmproxy<4.0.0; python_version < '3.6'
# https://github.com/pytest-dev/pytest-twisted/issues/93

531
tests/test_loader.py
View File

@ -1,14 +1,12 @@
from functools import partial
import unittest
import attr
from itemadapter import ItemAdapter
from itemloaders.processors import Compose, Identity, MapCompose, TakeFirst
from scrapy.http import HtmlResponse
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import (Compose, Identity, Join,
MapCompose, SelectJmes, TakeFirst)
from scrapy.selector import Selector
@ -69,6 +67,10 @@ def processor_with_args(value, other=None, loader_context=None):
class BasicItemLoaderTest(unittest.TestCase):
def test_add_value_on_unknown_field(self):
il = TestItemLoader()
self.assertRaises(KeyError, il.add_value, 'wrong_field', [u'lala', u'lolo'])
def test_load_item_using_default_loader(self):
i = TestItem()
i['summary'] = u'lala'
@ -85,391 +87,6 @@ class BasicItemLoaderTest(unittest.TestCase):
item = il.load_item()
self.assertEqual(item['name'], [u'Marta'])
def test_load_item_ignore_none_field_values(self):
def validate_sku(value):
# Let's assume a SKU is only digits.
if value.isdigit():
return value
class MyLoader(ItemLoader):
name_out = Compose(lambda vs: vs[0]) # take first which allows empty values
price_out = Compose(TakeFirst(), float)
sku_out = Compose(TakeFirst(), validate_sku)
valid_fragment = u'SKU: 1234'
invalid_fragment = u'SKU: not available'
sku_re = 'SKU: (.+)'
il = MyLoader(item={})
# Should not return "sku: None".
il.add_value('sku', [invalid_fragment], re=sku_re)
# Should not ignore empty values.
il.add_value('name', u'')
il.add_value('price', [u'0'])
self.assertEqual(il.load_item(), {
'name': u'',
'price': 0.0,
})
il.replace_value('sku', [valid_fragment], re=sku_re)
self.assertEqual(il.load_item()['sku'], u'1234')
def test_self_referencing_loader(self):
class MyLoader(ItemLoader):
url_out = TakeFirst()
def img_url_out(self, values):
return (self.get_output_value('url') or '') + values[0]
il = MyLoader(item={})
il.add_value('url', 'http://example.com/')
il.add_value('img_url', '1234.png')
self.assertEqual(il.load_item(), {
'url': 'http://example.com/',
'img_url': 'http://example.com/1234.png',
})
il = MyLoader(item={})
il.add_value('img_url', '1234.png')
self.assertEqual(il.load_item(), {
'img_url': '1234.png',
})
def test_add_value(self):
il = TestItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'Marta'])
self.assertEqual(il.get_output_value('name'), [u'Marta'])
il.add_value('name', u'pepe')
self.assertEqual(il.get_collected_values('name'), [u'Marta', u'Pepe'])
self.assertEqual(il.get_output_value('name'), [u'Marta', u'Pepe'])
# test add object value
il.add_value('summary', {'key': 1})
self.assertEqual(il.get_collected_values('summary'), [{'key': 1}])
il.add_value(None, u'Jim', lambda x: {'name': x})
self.assertEqual(il.get_collected_values('name'), [u'Marta', u'Pepe', u'Jim'])
def test_add_zero(self):
il = NameItemLoader()
il.add_value('name', 0)
self.assertEqual(il.get_collected_values('name'), [0])
def test_replace_value(self):
il = TestItemLoader()
il.replace_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'Marta'])
self.assertEqual(il.get_output_value('name'), [u'Marta'])
il.replace_value('name', u'pepe')
self.assertEqual(il.get_collected_values('name'), [u'Pepe'])
self.assertEqual(il.get_output_value('name'), [u'Pepe'])
il.replace_value(None, u'Jim', lambda x: {'name': x})
self.assertEqual(il.get_collected_values('name'), [u'Jim'])
def test_get_value(self):
il = NameItemLoader()
self.assertEqual(u'FOO', il.get_value([u'foo', u'bar'], TakeFirst(), str.upper))
self.assertEqual([u'foo', u'bar'], il.get_value([u'name:foo', u'name:bar'], re=u'name:(.*)$'))
self.assertEqual(u'foo', il.get_value([u'name:foo', u'name:bar'], TakeFirst(), re=u'name:(.*)$'))
il.add_value('name', [u'name:foo', u'name:bar'], TakeFirst(), re=u'name:(.*)$')
self.assertEqual([u'foo'], il.get_collected_values('name'))
il.replace_value('name', u'name:bar', re=u'name:(.*)$')
self.assertEqual([u'bar'], il.get_collected_values('name'))
def test_iter_on_input_processor_input(self):
class NameFirstItemLoader(NameItemLoader):
name_in = TakeFirst()
il = NameFirstItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.add_value('name', [u'marta', u'jose'])
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.replace_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.replace_value('name', [u'marta', u'jose'])
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.add_value('name', u'marta')
il.add_value('name', [u'jose', u'pedro'])
self.assertEqual(il.get_collected_values('name'), [u'marta', u'jose'])
def test_map_compose_filter(self):
def filter_world(x):
return None if x == 'world' else x
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc(['hello', 'world', 'this', 'is', 'scrapy']),
['HELLO', 'THIS', 'IS', 'SCRAPY'])
def test_map_compose_filter_multil(self):
class TestItemLoader(NameItemLoader):
name_in = MapCompose(lambda v: v.title(), lambda v: v[:-1])
il = TestItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Mart'])
item = il.load_item()
self.assertEqual(item['name'], [u'Mart'])
def test_default_input_processor(self):
il = DefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mart'])
def test_inherited_default_input_processor(self):
class InheritDefaultedItemLoader(DefaultedItemLoader):
pass
il = InheritDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mart'])
def test_input_processor_inheritance(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(lambda v: v.lower())
il = ChildItemLoader()
il.add_value('url', u'HTTP://scrapy.ORG')
self.assertEqual(il.get_output_value('url'), [u'http://scrapy.org'])
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Marta'])
class ChildChildItemLoader(ChildItemLoader):
url_in = MapCompose(lambda v: v.upper())
summary_in = MapCompose(lambda v: v)
il = ChildChildItemLoader()
il.add_value('url', u'http://scrapy.org')
self.assertEqual(il.get_output_value('url'), [u'HTTP://SCRAPY.ORG'])
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Marta'])
def test_empty_map_compose(self):
class IdentityDefaultedItemLoader(DefaultedItemLoader):
name_in = MapCompose()
il = IdentityDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'marta'])
def test_identity_input_processor(self):
class IdentityDefaultedItemLoader(DefaultedItemLoader):
name_in = Identity()
il = IdentityDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'marta'])
def test_extend_custom_input_processors(self):
class ChildItemLoader(TestItemLoader):
name_in = MapCompose(TestItemLoader.name_in, str.swapcase)
il = ChildItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mARTA'])
def test_extend_default_input_processors(self):
class ChildDefaultedItemLoader(DefaultedItemLoader):
name_in = MapCompose(DefaultedItemLoader.default_input_processor, str.swapcase)
il = ChildDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'MART'])
def test_output_processor_using_function(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class TakeFirstItemLoader(TestItemLoader):
name_out = u" ".join
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar Ta')
def test_output_processor_error(self):
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_out = MapCompose(float)
il = TestItemLoader()
il.add_value('name', [u'$10'])
try:
float(u'$10')
except Exception as e:
expected_exc_str = str(e)
exc = None
try:
il.load_item()
except Exception as e:
exc = e
assert isinstance(exc, ValueError)
s = str(exc)
assert 'name' in s, s
assert '$10' in s, s
assert 'ValueError' in s, s
assert expected_exc_str in s, s
def test_output_processor_using_classes(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class TakeFirstItemLoader(TestItemLoader):
name_out = Join()
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar Ta')
class TakeFirstItemLoader(TestItemLoader):
name_out = Join("<br>")
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar<br>Ta')
def test_default_output_processor(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class LalaItemLoader(TestItemLoader):
default_output_processor = Identity()
il = LalaItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
def test_loader_context_on_declaration(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args, key=u'val')
il = ChildItemLoader()
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_loader_context_on_instantiation(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args)
il = ChildItemLoader(key=u'val')
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_loader_context_on_assign(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args)
il = ChildItemLoader()
il.context['key'] = u'val'
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_item_passed_to_input_processor_functions(self):
def processor(value, loader_context):
return loader_context['item']['name']
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor)
it = TestItem(name='marta')
il = ChildItemLoader(item=it)
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['marta'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['marta'])
def test_add_value_on_unknown_field(self):
il = TestItemLoader()
self.assertRaises(KeyError, il.add_value, 'wrong_field', [u'lala', u'lolo'])
def test_compose_processor(self):
class TestItemLoader(NameItemLoader):
name_out = Compose(lambda v: v[0], lambda v: v.title(), lambda v: v[:-1])
il = TestItemLoader()
il.add_value('name', [u'marta', u'other'])
self.assertEqual(il.get_output_value('name'), u'Mart')
item = il.load_item()
self.assertEqual(item['name'], u'Mart')
def test_partial_processor(self):
def join(values, sep=None, loader_context=None, ignored=None):
if sep is not None:
return sep.join(values)
elif loader_context and 'sep' in loader_context:
return loader_context['sep'].join(values)
else:
return ''.join(values)
class TestItemLoader(NameItemLoader):
name_out = Compose(partial(join, sep='+'))
url_out = Compose(partial(join, loader_context={'sep': '.'}))
summary_out = Compose(partial(join, ignored='foo'))
il = TestItemLoader()
il.add_value('name', [u'rabbit', u'hole'])
il.add_value('url', [u'rabbit', u'hole'])
il.add_value('summary', [u'rabbit', u'hole'])
item = il.load_item()
self.assertEqual(item['name'], u'rabbit+hole')
self.assertEqual(item['url'], u'rabbit.hole')
self.assertEqual(item['summary'], u'rabbithole')
def test_error_input_processor(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_in = MapCompose(float)
il = TestItemLoader()
self.assertRaises(ValueError, il.add_value, 'name',
[u'marta', u'other'])
def test_error_output_processor(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_out = Compose(Join(), float)
il = TestItemLoader()
il.add_value('name', u'marta')
with self.assertRaises(ValueError):
il.load_item()
def test_error_processor_as_argument(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
il = TestItemLoader()
self.assertRaises(ValueError, il.add_value, 'name',
[u'marta', u'other'], Compose(float))
class InitializationTestMixin:
@ -587,41 +204,6 @@ class BaseNoInputReprocessingLoader(ItemLoader):
title_out = TakeFirst()
class NoInputReprocessingDictLoader(BaseNoInputReprocessingLoader):
default_item_class = dict
class NoInputReprocessingFromDictTest(unittest.TestCase):
"""
Loaders initialized from loaded items must not reprocess fields (dict instances)
"""
def test_avoid_reprocessing_with_initial_values_single(self):
il = NoInputReprocessingDictLoader(item=dict(title='foo'))
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='foo'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='foo'))
def test_avoid_reprocessing_with_initial_values_list(self):
il = NoInputReprocessingDictLoader(item=dict(title=['foo', 'bar']))
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='foo'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='foo'))
def test_avoid_reprocessing_without_initial_values_single(self):
il = NoInputReprocessingDictLoader()
il.add_value('title', 'foo')
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='FOO'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='FOO'))
def test_avoid_reprocessing_without_initial_values_list(self):
il = NoInputReprocessingDictLoader()
il.add_value('title', ['foo', 'bar'])
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='FOO'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='FOO'))
class NoInputReprocessingItem(Item):
title = Field()
@ -661,25 +243,6 @@ class NoInputReprocessingFromItemTest(unittest.TestCase):
self.assertEqual(NoInputReprocessingItemLoader(item=il_loaded).load_item(), {'title': 'FOO'})
class TestOutputProcessorDict(unittest.TestCase):
def test_output_processor(self):
class TempDict(dict):
def __init__(self, *args, **kwargs):
super(TempDict, self).__init__(self, *args, **kwargs)
self.setdefault('temp', 0.3)
class TempLoader(ItemLoader):
default_item_class = TempDict
default_input_processor = Identity()
default_output_processor = Compose(TakeFirst())
loader = TempLoader()
item = loader.load_item()
self.assertIsInstance(item, TempDict)
self.assertEqual(dict(item), {'temp': 0.3})
class TestOutputProcessorItem(unittest.TestCase):
def test_output_processor(self):
@ -701,49 +264,6 @@ class TestOutputProcessorItem(unittest.TestCase):
self.assertEqual(dict(item), {'temp': 0.3})
class ProcessorsTest(unittest.TestCase):
def test_take_first(self):
proc = TakeFirst()
self.assertEqual(proc([None, '', 'hello', 'world']), 'hello')
self.assertEqual(proc([None, '', 0, 'hello', 'world']), 0)
def test_identity(self):
proc = Identity()
self.assertEqual(proc([None, '', 'hello', 'world']),
[None, '', 'hello', 'world'])
def test_join(self):
proc = Join()
self.assertRaises(TypeError, proc, [None, '', 'hello', 'world'])
self.assertEqual(proc(['', 'hello', 'world']), u' hello world')
self.assertEqual(proc(['hello', 'world']), u'hello world')
self.assertIsInstance(proc(['hello', 'world']), str)
def test_compose(self):
proc = Compose(lambda v: v[0], str.upper)
self.assertEqual(proc(['hello', 'world']), 'HELLO')
proc = Compose(str.upper)
self.assertEqual(proc(None), None)
proc = Compose(str.upper, stop_on_none=False)
self.assertRaises(ValueError, proc, None)
proc = Compose(str.upper, lambda x: x + 1)
self.assertRaises(ValueError, proc, 'hello')
def test_mapcompose(self):
def filter_world(x):
return None if x == 'world' else x
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc([u'hello', u'world', u'this', u'is', u'scrapy']),
[u'HELLO', u'THIS', u'IS', u'SCRAPY'])
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc(None), [])
proc = MapCompose(filter_world, str.upper)
self.assertRaises(ValueError, proc, [1])
proc = MapCompose(filter_world, lambda x: x + 1)
self.assertRaises(ValueError, proc, 'hello')
class SelectortemLoaderTest(unittest.TestCase):
response = HtmlResponse(url="", encoding='utf-8', body=b"""
<html>
@ -921,6 +441,7 @@ class SubselectorLoaderTest(unittest.TestCase):
def test_nested_xpath(self):
l = NestedItemLoader(response=self.response)
nl = l.nested_xpath("//header")
nl.add_xpath('name', 'div/text()')
nl.add_css('name_div', '#id')
@ -998,31 +519,6 @@ class SubselectorLoaderTest(unittest.TestCase):
self.assertEqual(item['image'], [u'/images/logo.png'])
class SelectJmesTestCase(unittest.TestCase):
test_list_equals = {
'simple': ('foo.bar', {"foo": {"bar": "baz"}}, "baz"),
'invalid': ('foo.bar.baz', {"foo": {"bar": "baz"}}, None),
'top_level': ('foo', {"foo": {"bar": "baz"}}, {"bar": "baz"}),
'double_vs_single_quote_string': ('foo.bar', {"foo": {"bar": "baz"}}, "baz"),
'dict': (
'foo.bar[*].name',
{"foo": {"bar": [{"name": "one"}, {"name": "two"}]}},
['one', 'two']
),
'list': ('[1]', [1, 2], 2)
}
def test_output(self):
for l in self.test_list_equals:
expr, test_list, expected = self.test_list_equals[l]
test = SelectJmes(expr)(test_list)
self.assertEqual(
test,
expected,
msg='test "{}" got {} expected {}'.format(l, test, expected)
)
# Functions as processors
def function_processor_strip(iterable):
@ -1044,12 +540,6 @@ class FunctionProcessorItemLoader(ItemLoader):
default_item_class = FunctionProcessorItem
class FunctionProcessorDictLoader(ItemLoader):
default_item_class = dict
foo_in = function_processor_strip
foo_out = function_processor_upper
class FunctionProcessorTestCase(unittest.TestCase):
def test_processor_defined_in_item(self):
@ -1061,15 +551,6 @@ class FunctionProcessorTestCase(unittest.TestCase):
{'foo': ['BAR', 'ASDF', 'QWERTY']}
)
def test_processor_defined_in_item_loader(self):
lo = FunctionProcessorDictLoader()
lo.add_value('foo', ' bar ')
lo.add_value('foo', [' asdf ', ' qwerty '])
self.assertEqual(
dict(lo.load_item()),
{'foo': ['BAR', 'ASDF', 'QWERTY']}
)
if __name__ == "__main__":
unittest.main()

720
tests/test_loader_deprecated.py Normal file
View File

@ -0,0 +1,720 @@
"""
These tests are kept as references from the ones that were ported to a itemloaders library.
Once we remove the references from scrapy, we can remove these tests.
"""
import unittest
import warnings
from functools import partial
from itemloaders.processors import (Compose, Identity, Join,
MapCompose, SelectJmes, TakeFirst)
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.common import wrap_loader_context
from scrapy.utils.deprecate import ScrapyDeprecationWarning
from scrapy.utils.misc import extract_regex
# test items
class NameItem(Item):
name = Field()
class TestItem(NameItem):
url = Field()
summary = Field()
# test item loaders
class NameItemLoader(ItemLoader):
default_item_class = TestItem
class TestItemLoader(NameItemLoader):
name_in = MapCompose(lambda v: v.title())
class DefaultedItemLoader(NameItemLoader):
default_input_processor = MapCompose(lambda v: v[:-1])
# test processors
def processor_with_args(value, other=None, loader_context=None):
if 'key' in loader_context:
return loader_context['key']
return value
class BasicItemLoaderTest(unittest.TestCase):
def test_load_item_using_default_loader(self):
i = TestItem()
i['summary'] = u'lala'
il = ItemLoader(item=i)
il.add_value('name', u'marta')
item = il.load_item()
assert item is i
self.assertEqual(item['summary'], [u'lala'])
self.assertEqual(item['name'], [u'marta'])
def test_load_item_using_custom_loader(self):
il = TestItemLoader()
il.add_value('name', u'marta')
item = il.load_item()
self.assertEqual(item['name'], [u'Marta'])
def test_load_item_ignore_none_field_values(self):
def validate_sku(value):
# Let's assume a SKU is only digits.
if value.isdigit():
return value
class MyLoader(ItemLoader):
name_out = Compose(lambda vs: vs[0]) # take first which allows empty values
price_out = Compose(TakeFirst(), float)
sku_out = Compose(TakeFirst(), validate_sku)
valid_fragment = u'SKU: 1234'
invalid_fragment = u'SKU: not available'
sku_re = 'SKU: (.+)'
il = MyLoader(item={})
# Should not return "sku: None".
il.add_value('sku', [invalid_fragment], re=sku_re)
# Should not ignore empty values.
il.add_value('name', u'')
il.add_value('price', [u'0'])
self.assertEqual(il.load_item(), {
'name': u'',
'price': 0.0,
})
il.replace_value('sku', [valid_fragment], re=sku_re)
self.assertEqual(il.load_item()['sku'], u'1234')
def test_self_referencing_loader(self):
class MyLoader(ItemLoader):
url_out = TakeFirst()
def img_url_out(self, values):
return (self.get_output_value('url') or '') + values[0]
il = MyLoader(item={})
il.add_value('url', 'http://example.com/')
il.add_value('img_url', '1234.png')
self.assertEqual(il.load_item(), {
'url': 'http://example.com/',
'img_url': 'http://example.com/1234.png',
})
il = MyLoader(item={})
il.add_value('img_url', '1234.png')
self.assertEqual(il.load_item(), {
'img_url': '1234.png',
})
def test_add_value(self):
il = TestItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'Marta'])
self.assertEqual(il.get_output_value('name'), [u'Marta'])
il.add_value('name', u'pepe')
self.assertEqual(il.get_collected_values('name'), [u'Marta', u'Pepe'])
self.assertEqual(il.get_output_value('name'), [u'Marta', u'Pepe'])
# test add object value
il.add_value('summary', {'key': 1})
self.assertEqual(il.get_collected_values('summary'), [{'key': 1}])
il.add_value(None, u'Jim', lambda x: {'name': x})
self.assertEqual(il.get_collected_values('name'), [u'Marta', u'Pepe', u'Jim'])
def test_add_zero(self):
il = NameItemLoader()
il.add_value('name', 0)
self.assertEqual(il.get_collected_values('name'), [0])
def test_replace_value(self):
il = TestItemLoader()
il.replace_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'Marta'])
self.assertEqual(il.get_output_value('name'), [u'Marta'])
il.replace_value('name', u'pepe')
self.assertEqual(il.get_collected_values('name'), [u'Pepe'])
self.assertEqual(il.get_output_value('name'), [u'Pepe'])
il.replace_value(None, u'Jim', lambda x: {'name': x})
self.assertEqual(il.get_collected_values('name'), [u'Jim'])
def test_get_value(self):
il = NameItemLoader()
self.assertEqual(u'FOO', il.get_value([u'foo', u'bar'], TakeFirst(), str.upper))
self.assertEqual([u'foo', u'bar'], il.get_value([u'name:foo', u'name:bar'], re=u'name:(.*)$'))
self.assertEqual(u'foo', il.get_value([u'name:foo', u'name:bar'], TakeFirst(), re=u'name:(.*)$'))
il.add_value('name', [u'name:foo', u'name:bar'], TakeFirst(), re=u'name:(.*)$')
self.assertEqual([u'foo'], il.get_collected_values('name'))
il.replace_value('name', u'name:bar', re=u'name:(.*)$')
self.assertEqual([u'bar'], il.get_collected_values('name'))
def test_iter_on_input_processor_input(self):
class NameFirstItemLoader(NameItemLoader):
name_in = TakeFirst()
il = NameFirstItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.add_value('name', [u'marta', u'jose'])
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.replace_value('name', u'marta')
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.replace_value('name', [u'marta', u'jose'])
self.assertEqual(il.get_collected_values('name'), [u'marta'])
il = NameFirstItemLoader()
il.add_value('name', u'marta')
il.add_value('name', [u'jose', u'pedro'])
self.assertEqual(il.get_collected_values('name'), [u'marta', u'jose'])
def test_map_compose_filter(self):
def filter_world(x):
return None if x == 'world' else x
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc(['hello', 'world', 'this', 'is', 'scrapy']),
['HELLO', 'THIS', 'IS', 'SCRAPY'])
def test_map_compose_filter_multil(self):
class TestItemLoader(NameItemLoader):
name_in = MapCompose(lambda v: v.title(), lambda v: v[:-1])
il = TestItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Mart'])
item = il.load_item()
self.assertEqual(item['name'], [u'Mart'])
def test_default_input_processor(self):
il = DefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mart'])
def test_inherited_default_input_processor(self):
class InheritDefaultedItemLoader(DefaultedItemLoader):
pass
il = InheritDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mart'])
def test_input_processor_inheritance(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(lambda v: v.lower())
il = ChildItemLoader()
il.add_value('url', u'HTTP://scrapy.ORG')
self.assertEqual(il.get_output_value('url'), [u'http://scrapy.org'])
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Marta'])
class ChildChildItemLoader(ChildItemLoader):
url_in = MapCompose(lambda v: v.upper())
summary_in = MapCompose(lambda v: v)
il = ChildChildItemLoader()
il.add_value('url', u'http://scrapy.org')
self.assertEqual(il.get_output_value('url'), [u'HTTP://SCRAPY.ORG'])
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'Marta'])
def test_empty_map_compose(self):
class IdentityDefaultedItemLoader(DefaultedItemLoader):
name_in = MapCompose()
il = IdentityDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'marta'])
def test_identity_input_processor(self):
class IdentityDefaultedItemLoader(DefaultedItemLoader):
name_in = Identity()
il = IdentityDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'marta'])
def test_extend_custom_input_processors(self):
class ChildItemLoader(TestItemLoader):
name_in = MapCompose(TestItemLoader.name_in, str.swapcase)
il = ChildItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'mARTA'])
def test_extend_default_input_processors(self):
class ChildDefaultedItemLoader(DefaultedItemLoader):
name_in = MapCompose(DefaultedItemLoader.default_input_processor, str.swapcase)
il = ChildDefaultedItemLoader()
il.add_value('name', u'marta')
self.assertEqual(il.get_output_value('name'), [u'MART'])
def test_output_processor_using_function(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class TakeFirstItemLoader(TestItemLoader):
name_out = u" ".join
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar Ta')
def test_output_processor_error(self):
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_out = MapCompose(float)
il = TestItemLoader()
il.add_value('name', [u'$10'])
try:
float(u'$10')
except Exception as e:
expected_exc_str = str(e)
exc = None
try:
il.load_item()
except Exception as e:
exc = e
assert isinstance(exc, ValueError)
s = str(exc)
assert 'name' in s, s
assert '$10' in s, s
assert 'ValueError' in s, s
assert expected_exc_str in s, s
def test_output_processor_using_classes(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class TakeFirstItemLoader(TestItemLoader):
name_out = Join()
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar Ta')
class TakeFirstItemLoader(TestItemLoader):
name_out = Join("<br>")
il = TakeFirstItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), u'Mar<br>Ta')
def test_default_output_processor(self):
il = TestItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
class LalaItemLoader(TestItemLoader):
default_output_processor = Identity()
il = LalaItemLoader()
il.add_value('name', [u'mar', u'ta'])
self.assertEqual(il.get_output_value('name'), [u'Mar', u'Ta'])
def test_loader_context_on_declaration(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args, key=u'val')
il = ChildItemLoader()
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_loader_context_on_instantiation(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args)
il = ChildItemLoader(key=u'val')
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_loader_context_on_assign(self):
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor_with_args)
il = ChildItemLoader()
il.context['key'] = u'val'
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['val'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['val'])
def test_item_passed_to_input_processor_functions(self):
def processor(value, loader_context):
return loader_context['item']['name']
class ChildItemLoader(TestItemLoader):
url_in = MapCompose(processor)
it = TestItem(name='marta')
il = ChildItemLoader(item=it)
il.add_value('url', u'text')
self.assertEqual(il.get_output_value('url'), ['marta'])
il.replace_value('url', u'text2')
self.assertEqual(il.get_output_value('url'), ['marta'])
def test_compose_processor(self):
class TestItemLoader(NameItemLoader):
name_out = Compose(lambda v: v[0], lambda v: v.title(), lambda v: v[:-1])
il = TestItemLoader()
il.add_value('name', [u'marta', u'other'])
self.assertEqual(il.get_output_value('name'), u'Mart')
item = il.load_item()
self.assertEqual(item['name'], u'Mart')
def test_partial_processor(self):
def join(values, sep=None, loader_context=None, ignored=None):
if sep is not None:
return sep.join(values)
elif loader_context and 'sep' in loader_context:
return loader_context['sep'].join(values)
else:
return ''.join(values)
class TestItemLoader(NameItemLoader):
name_out = Compose(partial(join, sep='+'))
url_out = Compose(partial(join, loader_context={'sep': '.'}))
summary_out = Compose(partial(join, ignored='foo'))
il = TestItemLoader()
il.add_value('name', [u'rabbit', u'hole'])
il.add_value('url', [u'rabbit', u'hole'])
il.add_value('summary', [u'rabbit', u'hole'])
item = il.load_item()
self.assertEqual(item['name'], u'rabbit+hole')
self.assertEqual(item['url'], u'rabbit.hole')
self.assertEqual(item['summary'], u'rabbithole')
def test_error_input_processor(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_in = MapCompose(float)
il = TestItemLoader()
self.assertRaises(ValueError, il.add_value, 'name',
[u'marta', u'other'])
def test_error_output_processor(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
name_out = Compose(Join(), float)
il = TestItemLoader()
il.add_value('name', u'marta')
with self.assertRaises(ValueError):
il.load_item()
def test_error_processor_as_argument(self):
class TestItem(Item):
name = Field()
class TestItemLoader(ItemLoader):
default_item_class = TestItem
il = TestItemLoader()
self.assertRaises(ValueError, il.add_value, 'name',
[u'marta', u'other'], Compose(float))
class InitializationFromDictTest(unittest.TestCase):
item_class = dict
def test_keep_single_value(self):
"""Loaded item should contain values from the initial item"""
input_item = self.item_class(name='foo')
il = ItemLoader(item=input_item)
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo']})
def test_keep_list(self):
"""Loaded item should contain values from the initial item"""
input_item = self.item_class(name=['foo', 'bar'])
il = ItemLoader(item=input_item)
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo', 'bar']})
def test_add_value_singlevalue_singlevalue(self):
"""Values added after initialization should be appended"""
input_item = self.item_class(name='foo')
il = ItemLoader(item=input_item)
il.add_value('name', 'bar')
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo', 'bar']})
def test_add_value_singlevalue_list(self):
"""Values added after initialization should be appended"""
input_item = self.item_class(name='foo')
il = ItemLoader(item=input_item)
il.add_value('name', ['item', 'loader'])
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo', 'item', 'loader']})
def test_add_value_list_singlevalue(self):
"""Values added after initialization should be appended"""
input_item = self.item_class(name=['foo', 'bar'])
il = ItemLoader(item=input_item)
il.add_value('name', 'qwerty')
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo', 'bar', 'qwerty']})
def test_add_value_list_list(self):
"""Values added after initialization should be appended"""
input_item = self.item_class(name=['foo', 'bar'])
il = ItemLoader(item=input_item)
il.add_value('name', ['item', 'loader'])
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(dict(loaded_item), {'name': ['foo', 'bar', 'item', 'loader']})
def test_get_output_value_singlevalue(self):
"""Getting output value must not remove value from item"""
input_item = self.item_class(name='foo')
il = ItemLoader(item=input_item)
self.assertEqual(il.get_output_value('name'), ['foo'])
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(loaded_item, dict({'name': ['foo']}))
def test_get_output_value_list(self):
"""Getting output value must not remove value from item"""
input_item = self.item_class(name=['foo', 'bar'])
il = ItemLoader(item=input_item)
self.assertEqual(il.get_output_value('name'), ['foo', 'bar'])
loaded_item = il.load_item()
self.assertIsInstance(loaded_item, self.item_class)
self.assertEqual(loaded_item, dict({'name': ['foo', 'bar']}))
def test_values_single(self):
"""Values from initial item must be added to loader._values"""
input_item = self.item_class(name='foo')
il = ItemLoader(item=input_item)
self.assertEqual(il._values.get('name'), ['foo'])
def test_values_list(self):
"""Values from initial item must be added to loader._values"""
input_item = self.item_class(name=['foo', 'bar'])
il = ItemLoader(item=input_item)
self.assertEqual(il._values.get('name'), ['foo', 'bar'])
class BaseNoInputReprocessingLoader(ItemLoader):
title_in = MapCompose(str.upper)
title_out = TakeFirst()
class NoInputReprocessingDictLoader(BaseNoInputReprocessingLoader):
default_item_class = dict
class NoInputReprocessingFromDictTest(unittest.TestCase):
"""
Loaders initialized from loaded items must not reprocess fields (dict instances)
"""
def test_avoid_reprocessing_with_initial_values_single(self):
il = NoInputReprocessingDictLoader(item=dict(title='foo'))
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='foo'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='foo'))
def test_avoid_reprocessing_with_initial_values_list(self):
il = NoInputReprocessingDictLoader(item=dict(title=['foo', 'bar']))
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='foo'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='foo'))
def test_avoid_reprocessing_without_initial_values_single(self):
il = NoInputReprocessingDictLoader()
il.add_value('title', 'foo')
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='FOO'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='FOO'))
def test_avoid_reprocessing_without_initial_values_list(self):
il = NoInputReprocessingDictLoader()
il.add_value('title', ['foo', 'bar'])
il_loaded = il.load_item()
self.assertEqual(il_loaded, dict(title='FOO'))
self.assertEqual(NoInputReprocessingDictLoader(item=il_loaded).load_item(), dict(title='FOO'))
class TestOutputProcessorDict(unittest.TestCase):
def test_output_processor(self):
class TempDict(dict):
def __init__(self, *args, **kwargs):
super(TempDict, self).__init__(self, *args, **kwargs)
self.setdefault('temp', 0.3)
class TempLoader(ItemLoader):
default_item_class = TempDict
default_input_processor = Identity()
default_output_processor = Compose(TakeFirst())
loader = TempLoader()
item = loader.load_item()
self.assertIsInstance(item, TempDict)
self.assertEqual(dict(item), {'temp': 0.3})
class ProcessorsTest(unittest.TestCase):
def test_take_first(self):
proc = TakeFirst()
self.assertEqual(proc([None, '', 'hello', 'world']), 'hello')
self.assertEqual(proc([None, '', 0, 'hello', 'world']), 0)
def test_identity(self):
proc = Identity()
self.assertEqual(proc([None, '', 'hello', 'world']),
[None, '', 'hello', 'world'])
def test_join(self):
proc = Join()
self.assertRaises(TypeError, proc, [None, '', 'hello', 'world'])
self.assertEqual(proc(['', 'hello', 'world']), u' hello world')
self.assertEqual(proc(['hello', 'world']), u'hello world')
self.assertIsInstance(proc(['hello', 'world']), str)
def test_compose(self):
proc = Compose(lambda v: v[0], str.upper)
self.assertEqual(proc(['hello', 'world']), 'HELLO')
proc = Compose(str.upper)
self.assertEqual(proc(None), None)
proc = Compose(str.upper, stop_on_none=False)
self.assertRaises(ValueError, proc, None)
proc = Compose(str.upper, lambda x: x + 1)
self.assertRaises(ValueError, proc, 'hello')
def test_mapcompose(self):
def filter_world(x):
return None if x == 'world' else x
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc([u'hello', u'world', u'this', u'is', u'scrapy']),
[u'HELLO', u'THIS', u'IS', u'SCRAPY'])
proc = MapCompose(filter_world, str.upper)
self.assertEqual(proc(None), [])
proc = MapCompose(filter_world, str.upper)
self.assertRaises(ValueError, proc, [1])
proc = MapCompose(filter_world, lambda x: x + 1)
self.assertRaises(ValueError, proc, 'hello')
class SelectJmesTestCase(unittest.TestCase):
test_list_equals = {
'simple': ('foo.bar', {"foo": {"bar": "baz"}}, "baz"),
'invalid': ('foo.bar.baz', {"foo": {"bar": "baz"}}, None),
'top_level': ('foo', {"foo": {"bar": "baz"}}, {"bar": "baz"}),
'double_vs_single_quote_string': ('foo.bar', {"foo": {"bar": "baz"}}, "baz"),
'dict': (
'foo.bar[*].name',
{"foo": {"bar": [{"name": "one"}, {"name": "two"}]}},
['one', 'two']
),
'list': ('[1]', [1, 2], 2)
}
def test_output(self):
for tl in self.test_list_equals:
expr, test_list, expected = self.test_list_equals[tl]
test = SelectJmes(expr)(test_list)
self.assertEqual(
test,
expected,
msg='test "{}" got {} expected {}'.format(tl, test, expected)
)
# Functions as processors
def function_processor_strip(iterable):
return [x.strip() for x in iterable]
def function_processor_upper(iterable):
return [x.upper() for x in iterable]
class FunctionProcessorItem(Item):
foo = Field(
input_processor=function_processor_strip,
output_processor=function_processor_upper,
)
class FunctionProcessorDictLoader(ItemLoader):
default_item_class = dict
foo_in = function_processor_strip
foo_out = function_processor_upper
class FunctionProcessorTestCase(unittest.TestCase):
def test_processor_defined_in_item_loader(self):
lo = FunctionProcessorDictLoader()
lo.add_value('foo', ' bar ')
lo.add_value('foo', [' asdf ', ' qwerty '])
self.assertEqual(
dict(lo.load_item()),
{'foo': ['BAR', 'ASDF', 'QWERTY']}
)
class DeprecatedUtilityFunctionsTestCase(unittest.TestCase):
def test_deprecated_wrap_loader_context(self):
def function(*args):
return None
with warnings.catch_warnings(record=True) as w:
wrap_loader_context(function, context=dict())
assert len(w) == 1
assert issubclass(w[0].category, ScrapyDeprecationWarning)
def test_deprecated_extract_regex(self):
with warnings.catch_warnings(record=True) as w:
extract_regex(r'\w+', 'this is a test')
assert len(w) == 1
assert issubclass(w[0].category, ScrapyDeprecationWarning)
if __name__ == "__main__":
unittest.main()