Skip to content
Snippets Groups Projects
Verified Commit f6e19bf4 authored by Gaëtan Blond's avatar Gaëtan Blond
Browse files

Added Atom RSS fetcher

parent d49b8610
No related branches found
No related tags found
1 merge request!1Refactor and clean old code
import logging
from typing import List
import defusedxml.ElementTree as ET
import requests
from ..entry import Entry
from .project_fetcher import ProjectFetcher
logger = logging.getLogger(__name__)
_ATOM_STANDARD = "{http://www.w3.org/2005/Atom}"
_ATOM_ENTRY = f"{_ATOM_STANDARD}entry"
_ATOM_TAG_ID = f"{_ATOM_STANDARD}id"
_ATOM_TAG_VERSION = f"{_ATOM_STANDARD}title"
_ATOM_TAG_URL = f"{_ATOM_STANDARD}link"
class AtomRSSFetcher(ProjectFetcher):
def __init__(self, feed_id: int, name: str, stream_url: str):
super().__init__(feed_id, name)
self.stream_url = stream_url
def fetch_entries(self) -> List[Entry]:
req = requests.get(self.stream_url)
if req.status_code != requests.codes.ok:
logger.error(
"Failed to get %s RSS file at %s, server returned %s.",
self.name,
self.stream_url,
req.status_code,
)
return []
entries: List[Entry] = []
root = ET.fromstring(req.content)
ET.fromstring(req.content)
for entry in root.findall(_ATOM_ENTRY):
tag_id = entry.findtext(_ATOM_TAG_ID)
version = entry.findtext(_ATOM_TAG_VERSION)
url = entry.find(_ATOM_TAG_URL)
entries.append(Entry(self.feed_id, tag_id, self.name, version, url))
return entries
import logging
import requests
from typing import List
import requests
from ..entry import Entry
from .project_fetcher import ProjectFetcher
......
import logging
import requests
from typing import List
import urllib.parse
import requests
from ..entry import Entry
from .project_fetcher import ProjectFetcher
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment