mirror of https://github.com/scrapy/scrapy.git
Full typing for scrapy/exporters.py. (#6275)
This commit is contained in:
Andrey Rakhmatullin
GitHub
parent
861646fbb3
commit
8985a04bd1
|
|
@ -3,16 +3,18 @@ Item Exporters are used to export/serialize items into different formats.
|
|||
"""
|
||||
|
||||
import csv
|
||||
import io
|
||||
import marshal
|
||||
import pickle # nosec
|
||||
import pprint
|
||||
from collections.abc import Mapping
|
||||
from io import BytesIO, TextIOWrapper
|
||||
from json import JSONEncoder
|
||||
from typing import Any, Callable, Dict, Iterable, Mapping, Optional, Tuple, Union
|
||||
from xml.sax.saxutils import XMLGenerator # nosec
|
||||
from xml.sax.xmlreader import AttributesImpl # nosec
|
||||
|
||||
from itemadapter import ItemAdapter, is_item
|
||||
|
||||
from scrapy.item import Item
|
||||
from scrapy.item import Field, Item
|
||||
from scrapy.utils.python import is_listlike, to_bytes, to_unicode
|
||||
from scrapy.utils.serialize import ScrapyJSONEncoder
|
||||
|
||||
|
|
@ -29,36 +31,42 @@ __all__ = [
|
|||
|
||||
|
||||
class BaseItemExporter:
|
||||
def __init__(self, *, dont_fail=False, **kwargs):
|
||||
self._kwargs = kwargs
|
||||
def __init__(self, *, dont_fail: bool = False, **kwargs: Any):
|
||||
self._kwargs: Dict[str, Any] = kwargs
|
||||
self._configure(kwargs, dont_fail=dont_fail)
|
||||
|
||||
def _configure(self, options, dont_fail=False):
|
||||
def _configure(self, options: Dict[str, Any], dont_fail: bool = False) -> None:
|
||||
"""Configure the exporter by popping options from the ``options`` dict.
|
||||
If dont_fail is set, it won't raise an exception on unexpected options
|
||||
(useful for using with keyword arguments in subclasses ``__init__`` methods)
|
||||
"""
|
||||
self.encoding = options.pop("encoding", None)
|
||||
self.fields_to_export = options.pop("fields_to_export", None)
|
||||
self.export_empty_fields = options.pop("export_empty_fields", False)
|
||||
self.indent = options.pop("indent", None)
|
||||
self.encoding: Optional[str] = options.pop("encoding", None)
|
||||
self.fields_to_export: Union[Mapping[str, str], Iterable[str], None] = (
|
||||
options.pop("fields_to_export", None)
|
||||
)
|
||||
self.export_empty_fields: bool = options.pop("export_empty_fields", False)
|
||||
self.indent: Optional[int] = options.pop("indent", None)
|
||||
if not dont_fail and options:
|
||||
raise TypeError(f"Unexpected options: {', '.join(options.keys())}")
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def serialize_field(self, field, name, value):
|
||||
serializer = field.get("serializer", lambda x: x)
|
||||
def serialize_field(
|
||||
self, field: Union[Mapping[str, Any], Field], name: str, value: Any
|
||||
) -> Any:
|
||||
serializer: Callable[[Any], Any] = field.get("serializer", lambda x: x)
|
||||
return serializer(value)
|
||||
|
||||
def start_exporting(self):
|
||||
def start_exporting(self) -> None:
|
||||
pass
|
||||
|
||||
def finish_exporting(self):
|
||||
def finish_exporting(self) -> None:
|
||||
pass
|
||||
|
||||
def _get_serialized_fields(self, item, default_value=None, include_empty=None):
|
||||
def _get_serialized_fields(
|
||||
self, item: Any, default_value: Any = None, include_empty: Optional[bool] = None
|
||||
) -> Iterable[Tuple[str, Any]]:
|
||||
"""Return the fields to export as an iterable of tuples
|
||||
(name, serialized_value)
|
||||
"""
|
||||
|
|
@ -100,22 +108,22 @@ class BaseItemExporter:
|
|||
|
||||
|
||||
class JsonLinesItemExporter(BaseItemExporter):
|
||||
def __init__(self, file, **kwargs):
|
||||
def __init__(self, file: BytesIO, **kwargs: Any):
|
||||
super().__init__(dont_fail=True, **kwargs)
|
||||
self.file = file
|
||||
self.file: BytesIO = file
|
||||
self._kwargs.setdefault("ensure_ascii", not self.encoding)
|
||||
self.encoder = ScrapyJSONEncoder(**self._kwargs)
|
||||
self.encoder: JSONEncoder = ScrapyJSONEncoder(**self._kwargs)
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
itemdict = dict(self._get_serialized_fields(item))
|
||||
data = self.encoder.encode(itemdict) + "\n"
|
||||
self.file.write(to_bytes(data, self.encoding))
|
||||
|
||||
|
||||
class JsonItemExporter(BaseItemExporter):
|
||||
def __init__(self, file, **kwargs):
|
||||
def __init__(self, file: BytesIO, **kwargs: Any):
|
||||
super().__init__(dont_fail=True, **kwargs)
|
||||
self.file = file
|
||||
self.file: BytesIO = file
|
||||
# there is a small difference between the behaviour or JsonItemExporter.indent
|
||||
# and ScrapyJSONEncoder.indent. ScrapyJSONEncoder.indent=None is needed to prevent
|
||||
# the addition of newlines everywhere
|
||||
|
|
@ -127,26 +135,26 @@ class JsonItemExporter(BaseItemExporter):
|
|||
self.encoder = ScrapyJSONEncoder(**self._kwargs)
|
||||
self.first_item = True
|
||||
|
||||
def _beautify_newline(self):
|
||||
def _beautify_newline(self) -> None:
|
||||
if self.indent is not None:
|
||||
self.file.write(b"\n")
|
||||
|
||||
def _add_comma_after_first(self):
|
||||
def _add_comma_after_first(self) -> None:
|
||||
if self.first_item:
|
||||
self.first_item = False
|
||||
else:
|
||||
self.file.write(b",")
|
||||
self._beautify_newline()
|
||||
|
||||
def start_exporting(self):
|
||||
def start_exporting(self) -> None:
|
||||
self.file.write(b"[")
|
||||
self._beautify_newline()
|
||||
|
||||
def finish_exporting(self):
|
||||
def finish_exporting(self) -> None:
|
||||
self._beautify_newline()
|
||||
self.file.write(b"]")
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
itemdict = dict(self._get_serialized_fields(item))
|
||||
data = to_bytes(self.encoder.encode(itemdict), self.encoding)
|
||||
self._add_comma_after_first()
|
||||
|
|
@ -154,7 +162,7 @@ class JsonItemExporter(BaseItemExporter):
|
|||
|
||||
|
||||
class XmlItemExporter(BaseItemExporter):
|
||||
def __init__(self, file, **kwargs):
|
||||
def __init__(self, file: BytesIO, **kwargs: Any):
|
||||
self.item_element = kwargs.pop("item_element", "item")
|
||||
self.root_element = kwargs.pop("root_element", "items")
|
||||
super().__init__(**kwargs)
|
||||
|
|
@ -162,22 +170,22 @@ class XmlItemExporter(BaseItemExporter):
|
|||
self.encoding = "utf-8"
|
||||
self.xg = XMLGenerator(file, encoding=self.encoding)
|
||||
|
||||
def _beautify_newline(self, new_item=False):
|
||||
def _beautify_newline(self, new_item: bool = False) -> None:
|
||||
if self.indent is not None and (self.indent > 0 or new_item):
|
||||
self.xg.characters("\n")
|
||||
|
||||
def _beautify_indent(self, depth=1):
|
||||
def _beautify_indent(self, depth: int = 1) -> None:
|
||||
if self.indent:
|
||||
self.xg.characters(" " * self.indent * depth)
|
||||
|
||||
def start_exporting(self):
|
||||
def start_exporting(self) -> None:
|
||||
self.xg.startDocument()
|
||||
self.xg.startElement(self.root_element, {})
|
||||
self.xg.startElement(self.root_element, AttributesImpl({}))
|
||||
self._beautify_newline(new_item=True)
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
self._beautify_indent(depth=1)
|
||||
self.xg.startElement(self.item_element, {})
|
||||
self.xg.startElement(self.item_element, AttributesImpl({}))
|
||||
self._beautify_newline()
|
||||
for name, value in self._get_serialized_fields(item, default_value=""):
|
||||
self._export_xml_field(name, value, depth=2)
|
||||
|
|
@ -185,13 +193,13 @@ class XmlItemExporter(BaseItemExporter):
|
|||
self.xg.endElement(self.item_element)
|
||||
self._beautify_newline(new_item=True)
|
||||
|
||||
def finish_exporting(self):
|
||||
def finish_exporting(self) -> None:
|
||||
self.xg.endElement(self.root_element)
|
||||
self.xg.endDocument()
|
||||
|
||||
def _export_xml_field(self, name, serialized_value, depth):
|
||||
def _export_xml_field(self, name: str, serialized_value: Any, depth: int) -> None:
|
||||
self._beautify_indent(depth=depth)
|
||||
self.xg.startElement(name, {})
|
||||
self.xg.startElement(name, AttributesImpl({}))
|
||||
if hasattr(serialized_value, "items"):
|
||||
self._beautify_newline()
|
||||
for subname, value in serialized_value.items():
|
||||
|
|
@ -213,17 +221,17 @@ class XmlItemExporter(BaseItemExporter):
|
|||
class CsvItemExporter(BaseItemExporter):
|
||||
def __init__(
|
||||
self,
|
||||
file,
|
||||
include_headers_line=True,
|
||||
join_multivalued=",",
|
||||
errors=None,
|
||||
**kwargs,
|
||||
file: BytesIO,
|
||||
include_headers_line: bool = True,
|
||||
join_multivalued: str = ",",
|
||||
errors: Optional[str] = None,
|
||||
**kwargs: Any,
|
||||
):
|
||||
super().__init__(dont_fail=True, **kwargs)
|
||||
if not self.encoding:
|
||||
self.encoding = "utf-8"
|
||||
self.include_headers_line = include_headers_line
|
||||
self.stream = io.TextIOWrapper(
|
||||
self.stream = TextIOWrapper(
|
||||
file,
|
||||
line_buffering=False,
|
||||
write_through=True,
|
||||
|
|
@ -235,11 +243,13 @@ class CsvItemExporter(BaseItemExporter):
|
|||
self._headers_not_written = True
|
||||
self._join_multivalued = join_multivalued
|
||||
|
||||
def serialize_field(self, field, name, value):
|
||||
serializer = field.get("serializer", self._join_if_needed)
|
||||
def serialize_field(
|
||||
self, field: Union[Mapping[str, Any], Field], name: str, value: Any
|
||||
) -> Any:
|
||||
serializer: Callable[[Any], Any] = field.get("serializer", self._join_if_needed)
|
||||
return serializer(value)
|
||||
|
||||
def _join_if_needed(self, value):
|
||||
def _join_if_needed(self, value: Any) -> Any:
|
||||
if isinstance(value, (list, tuple)):
|
||||
try:
|
||||
return self._join_multivalued.join(value)
|
||||
|
|
@ -247,7 +257,7 @@ class CsvItemExporter(BaseItemExporter):
|
|||
pass
|
||||
return value
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
if self._headers_not_written:
|
||||
self._headers_not_written = False
|
||||
self._write_headers_and_set_fields_to_export(item)
|
||||
|
|
@ -256,36 +266,38 @@ class CsvItemExporter(BaseItemExporter):
|
|||
values = list(self._build_row(x for _, x in fields))
|
||||
self.csv_writer.writerow(values)
|
||||
|
||||
def finish_exporting(self):
|
||||
def finish_exporting(self) -> None:
|
||||
self.stream.detach() # Avoid closing the wrapped file.
|
||||
|
||||
def _build_row(self, values):
|
||||
def _build_row(self, values: Iterable[Any]) -> Iterable[Any]:
|
||||
for s in values:
|
||||
try:
|
||||
yield to_unicode(s, self.encoding)
|
||||
except TypeError:
|
||||
yield s
|
||||
|
||||
def _write_headers_and_set_fields_to_export(self, item):
|
||||
def _write_headers_and_set_fields_to_export(self, item: Any) -> None:
|
||||
if self.include_headers_line:
|
||||
if not self.fields_to_export:
|
||||
# use declared field names, or keys if the item is a dict
|
||||
self.fields_to_export = ItemAdapter(item).field_names()
|
||||
fields: Iterable[str]
|
||||
if isinstance(self.fields_to_export, Mapping):
|
||||
fields = self.fields_to_export.values()
|
||||
else:
|
||||
assert self.fields_to_export
|
||||
fields = self.fields_to_export
|
||||
row = list(self._build_row(fields))
|
||||
self.csv_writer.writerow(row)
|
||||
|
||||
|
||||
class PickleItemExporter(BaseItemExporter):
|
||||
def __init__(self, file, protocol=4, **kwargs):
|
||||
def __init__(self, file: BytesIO, protocol: int = 4, **kwargs: Any):
|
||||
super().__init__(**kwargs)
|
||||
self.file = file
|
||||
self.protocol = protocol
|
||||
self.file: BytesIO = file
|
||||
self.protocol: int = protocol
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
d = dict(self._get_serialized_fields(item))
|
||||
pickle.dump(d, self.file, self.protocol)
|
||||
|
||||
|
|
@ -299,20 +311,20 @@ class MarshalItemExporter(BaseItemExporter):
|
|||
opened in binary mode, a :class:`~io.BytesIO` object, etc)
|
||||
"""
|
||||
|
||||
def __init__(self, file, **kwargs):
|
||||
def __init__(self, file: BytesIO, **kwargs: Any):
|
||||
super().__init__(**kwargs)
|
||||
self.file = file
|
||||
self.file: BytesIO = file
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
marshal.dump(dict(self._get_serialized_fields(item)), self.file)
|
||||
|
||||
|
||||
class PprintItemExporter(BaseItemExporter):
|
||||
def __init__(self, file, **kwargs):
|
||||
def __init__(self, file: BytesIO, **kwargs: Any):
|
||||
super().__init__(**kwargs)
|
||||
self.file = file
|
||||
self.file: BytesIO = file
|
||||
|
||||
def export_item(self, item):
|
||||
def export_item(self, item: Any) -> None:
|
||||
itemdict = dict(self._get_serialized_fields(item))
|
||||
self.file.write(to_bytes(pprint.pformat(itemdict) + "\n"))
|
||||
|
||||
|
|
@ -327,16 +339,20 @@ class PythonItemExporter(BaseItemExporter):
|
|||
.. _msgpack: https://pypi.org/project/msgpack/
|
||||
"""
|
||||
|
||||
def _configure(self, options, dont_fail=False):
|
||||
def _configure(self, options: Dict[str, Any], dont_fail: bool = False) -> None:
|
||||
super()._configure(options, dont_fail)
|
||||
if not self.encoding:
|
||||
self.encoding = "utf-8"
|
||||
|
||||
def serialize_field(self, field, name, value):
|
||||
serializer = field.get("serializer", self._serialize_value)
|
||||
def serialize_field(
|
||||
self, field: Union[Mapping[str, Any], Field], name: str, value: Any
|
||||
) -> Any:
|
||||
serializer: Callable[[Any], Any] = field.get(
|
||||
"serializer", self._serialize_value
|
||||
)
|
||||
return serializer(value)
|
||||
|
||||
def _serialize_value(self, value):
|
||||
def _serialize_value(self, value: Any) -> Any:
|
||||
if isinstance(value, Item):
|
||||
return self.export_item(value)
|
||||
if is_item(value):
|
||||
|
|
@ -347,10 +363,10 @@ class PythonItemExporter(BaseItemExporter):
|
|||
return to_unicode(value, encoding=self.encoding)
|
||||
return value
|
||||
|
||||
def _serialize_item(self, item):
|
||||
def _serialize_item(self, item: Any) -> Iterable[Tuple[Union[str, bytes], Any]]:
|
||||
for key, value in ItemAdapter(item).items():
|
||||
yield key, self._serialize_value(value)
|
||||
|
||||
def export_item(self, item):
|
||||
result = dict(self._get_serialized_fields(item))
|
||||
def export_item(self, item: Any) -> Dict[Union[str, bytes], Any]: # type: ignore[override]
|
||||
result: Dict[Union[str, bytes], Any] = dict(self._get_serialized_fields(item))
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ import warnings
|
|||
from datetime import datetime, timezone
|
||||
from pathlib import Path, PureWindowsPath
|
||||
from tempfile import NamedTemporaryFile
|
||||
from typing import IO, Any, Callable, Dict, List, Optional, Tuple, Union
|
||||
from typing import IO, Any, Callable, Dict, List, Optional, Tuple, Type, Union
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
from twisted.internet import defer, threads
|
||||
|
|
@ -21,6 +21,7 @@ from zope.interface import Interface, implementer
|
|||
|
||||
from scrapy import Spider, signals
|
||||
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
|
||||
from scrapy.exporters import BaseItemExporter
|
||||
from scrapy.extensions.postprocessing import PostProcessingManager
|
||||
from scrapy.utils.boto import is_botocore_available
|
||||
from scrapy.utils.conf import feed_complete_default_values_from_settings
|
||||
|
|
@ -324,12 +325,12 @@ class FeedSlot:
|
|||
filter,
|
||||
feed_options,
|
||||
spider,
|
||||
exporters,
|
||||
exporters: Dict[str, Type[BaseItemExporter]],
|
||||
settings,
|
||||
crawler,
|
||||
):
|
||||
self.file = None
|
||||
self.exporter = None
|
||||
self.exporter: Optional[BaseItemExporter] = None
|
||||
self.storage = storage
|
||||
# feed params
|
||||
self.batch_id = batch_id
|
||||
|
|
@ -341,7 +342,7 @@ class FeedSlot:
|
|||
# exporter params
|
||||
self.feed_options = feed_options
|
||||
self.spider = spider
|
||||
self.exporters = exporters
|
||||
self.exporters: Dict[str, Type[BaseItemExporter]] = exporters
|
||||
self.settings = settings
|
||||
self.crawler = crawler
|
||||
# flags
|
||||
|
|
@ -373,7 +374,7 @@ class FeedSlot:
|
|||
def _get_instance(self, objcls, *args, **kwargs):
|
||||
return build_from_crawler(objcls, self.crawler, *args, **kwargs)
|
||||
|
||||
def _get_exporter(self, file, format, *args, **kwargs):
|
||||
def _get_exporter(self, file, format, *args, **kwargs) -> BaseItemExporter:
|
||||
return self._get_instance(self.exporters[format], file, *args, **kwargs)
|
||||
|
||||
def finish_exporting(self):
|
||||
|
|
|
|||
Loading…
Reference in New Issue