|
1 |
| -from typing import Optional, Type |
| 1 | +from typing import ClassVar, Optional, Type |
2 | 2 |
|
3 | 3 | from xmltodict import parse
|
4 | 4 |
|
| 5 | +from rss_parser.custom_decorators import abstract_class_attributes |
5 | 6 | from rss_parser.models import XMLBaseModel
|
6 | 7 | from rss_parser.models.atom import Atom
|
7 | 8 | from rss_parser.models.rss import RSS
|
|
14 | 15 | # TODO: Older RSS versions
|
15 | 16 |
|
16 | 17 |
|
17 |
| -class Parser: |
| 18 | +@abstract_class_attributes("schema") |
| 19 | +class BaseParser: |
18 | 20 | """Parser for rss/atom files."""
|
19 | 21 |
|
20 |
| - @staticmethod |
21 |
| - def check_schema(root: dict) -> tuple[dict, type[XMLBaseModel]]: |
22 |
| - if "feed" in root: |
23 |
| - return root, Atom |
24 |
| - return root["rss"], RSS |
| 22 | + schema: ClassVar[Type[XMLBaseModel]] |
| 23 | + root_key: Optional[str] = None |
25 | 24 |
|
26 | 25 | @staticmethod
|
27 | 26 | def to_xml(data: str, *args, **kwargs):
|
28 | 27 | return parse(str(data), *args, **kwargs)
|
29 | 28 |
|
30 | 29 | @classmethod
|
31 |
| - def parse(cls, data: str, *, schema: Optional[Type[XMLBaseModel]] = None, root_key: str = "") -> XMLBaseModel: |
| 30 | + def parse( |
| 31 | + cls, |
| 32 | + data: str, |
| 33 | + *, |
| 34 | + schema: Optional[Type[XMLBaseModel]] = None, |
| 35 | + root_key: Optional[str] = None, |
| 36 | + ) -> XMLBaseModel: |
32 | 37 | """
|
33 |
| - Parse XML data into schema (default: RSS 2.0 or Atom). |
34 |
| -
|
| 38 | + Parse XML data into schema. |
35 | 39 | :param data: string of XML data that needs to be parsed
|
36 | 40 | :return: "schema" object
|
37 | 41 | """
|
38 | 42 | root = cls.to_xml(data)
|
39 |
| - if not isinstance(schema, XMLBaseModel): |
40 |
| - root, schema = cls.check_schema(root) |
41 |
| - else: |
| 43 | + |
| 44 | + schema = schema if schema else cls.schema |
| 45 | + |
| 46 | + root_key = root_key if root_key else cls.root_key |
| 47 | + |
| 48 | + if root_key: |
42 | 49 | root = root.get(root_key, root)
|
43 | 50 |
|
44 | 51 | return schema.parse_obj(root)
|
| 52 | + |
| 53 | + |
| 54 | +class AtomParser(BaseParser): |
| 55 | + schema = Atom |
| 56 | + |
| 57 | + |
| 58 | +class RSSParser(BaseParser): |
| 59 | + root_key = "rss" |
| 60 | + schema = RSS |
| 61 | + |
| 62 | + |
| 63 | +class Parser(RSSParser): |
| 64 | + @classmethod |
| 65 | + def parse(cls, data: str, *, schema: Optional[Type[XMLBaseModel]] = None) -> XMLBaseModel: |
| 66 | + import warnings |
| 67 | + |
| 68 | + warnings.warn( |
| 69 | + "Class Parser was renamed to RSSParser " "and will be removed in the next major update", |
| 70 | + DeprecationWarning, |
| 71 | + stacklevel=2, |
| 72 | + ) |
| 73 | + return RSSParser.parse(data, schema=schema) |
0 commit comments