refactor: make webspeiseplan fetching explicit
This commit is contained in:
@@ -1,9 +1,22 @@
|
|||||||
import logging
|
import logging
|
||||||
import urllib.request
|
import urllib.request
|
||||||
|
import urllib.parse
|
||||||
import re
|
import re
|
||||||
import time
|
import time
|
||||||
import json
|
import json
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class SWPWebspeiseplanData:
|
||||||
|
"""Downloaded SWP Webspeiseplan data grouped by outlet name."""
|
||||||
|
|
||||||
|
outlets: dict[str, dict]
|
||||||
|
locations: dict[str, dict]
|
||||||
|
menus: dict[str, dict]
|
||||||
|
meal_categories: dict[str, dict]
|
||||||
|
|
||||||
|
|
||||||
class SWPWebspeiseplanAPI:
|
class SWPWebspeiseplanAPI:
|
||||||
"""This class is used download content from SWP_Webspeiseplan.
|
"""This class is used download content from SWP_Webspeiseplan.
|
||||||
@@ -16,25 +29,34 @@ class SWPWebspeiseplanAPI:
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
"""Initialize the configuration for the web service."""
|
"""Initialize the web service client."""
|
||||||
logging.basicConfig()
|
logging.basicConfig()
|
||||||
|
|
||||||
|
def fetch_all(self) -> SWPWebspeiseplanData:
|
||||||
|
"""Download all data required to render OpenMensa feeds."""
|
||||||
proxy_token = self.parse_token()
|
proxy_token = self.parse_token()
|
||||||
self.outlets = self.parse_outlets(proxy_token)
|
outlets = self.parse_outlets(proxy_token)
|
||||||
self.locations: dict[str, dict] = {}
|
|
||||||
locations = {
|
locations = {
|
||||||
item["id"]: item
|
item["id"]: item
|
||||||
for item in self.parse_location(proxy_token)
|
for item in self.parse_location(proxy_token)
|
||||||
}
|
}
|
||||||
self.menus: dict[str, dict] = {}
|
menus: dict[str, dict] = {}
|
||||||
self.meal_categories: dict[str, dict] = {}
|
meal_categories: dict[str, dict] = {}
|
||||||
for outlet in self.outlets.values():
|
outlet_locations: dict[str, dict] = {}
|
||||||
|
for outlet in outlets.values():
|
||||||
location = outlet["standortID"]
|
location = outlet["standortID"]
|
||||||
menu = self.parse_menu(proxy_token, location)
|
menu = self.parse_menu(proxy_token, location)
|
||||||
categories = self.parse_meal_category(proxy_token, location)
|
categories = self.parse_meal_category(proxy_token, location)
|
||||||
id2cat = {item["gerichtkategorieID"]: item for item in categories}
|
id2cat = {item["gerichtkategorieID"]: item for item in categories}
|
||||||
self.menus[outlet["name"]] = menu
|
menus[outlet["name"]] = menu
|
||||||
self.meal_categories[outlet["name"]] = id2cat
|
meal_categories[outlet["name"]] = id2cat
|
||||||
self.locations[outlet["name"]] = locations[location]
|
outlet_locations[outlet["name"]] = locations[location]
|
||||||
|
return SWPWebspeiseplanData(
|
||||||
|
outlets=outlets,
|
||||||
|
locations=outlet_locations,
|
||||||
|
menus=menus,
|
||||||
|
meal_categories=meal_categories,
|
||||||
|
)
|
||||||
|
|
||||||
def __spoof_req_headers(self, req: urllib.request.Request):
|
def __spoof_req_headers(self, req: urllib.request.Request):
|
||||||
"""Add headers to a request .
|
"""Add headers to a request .
|
||||||
@@ -77,9 +99,8 @@ class SWPWebspeiseplanAPI:
|
|||||||
Returns:
|
Returns:
|
||||||
[type]: [description]
|
[type]: [description]
|
||||||
"""
|
"""
|
||||||
url = f"{SWPWebspeiseplanAPI.URL_BASE}/index.php?" + "&".join(
|
query = urllib.parse.urlencode(params)
|
||||||
[f"{k}={v}" for k, v in params.items()]
|
url = f"{SWPWebspeiseplanAPI.URL_BASE}/index.php?{query}"
|
||||||
)
|
|
||||||
SWPWebspeiseplanAPI.logger.debug("__parse_model: %s", url)
|
SWPWebspeiseplanAPI.logger.debug("__parse_model: %s", url)
|
||||||
req = urllib.request.Request(url)
|
req = urllib.request.Request(url)
|
||||||
self.__spoof_req_headers(req)
|
self.__spoof_req_headers(req)
|
||||||
|
|||||||
@@ -3,7 +3,10 @@ from dataclasses import dataclass
|
|||||||
import logging
|
import logging
|
||||||
from flask import url_for
|
from flask import url_for
|
||||||
from stw_potsdam.xml_types.openmensa_xml import OpenMensaXML
|
from stw_potsdam.xml_types.openmensa_xml import OpenMensaXML
|
||||||
from stw_potsdam.swp_webspeiseplan_api import SWPWebspeiseplanAPI
|
from stw_potsdam.swp_webspeiseplan_api import (
|
||||||
|
SWPWebspeiseplanAPI,
|
||||||
|
SWPWebspeiseplanData,
|
||||||
|
)
|
||||||
from stw_potsdam.swp_webspeiseplan_parser import SWPWebspeiseplanParser
|
from stw_potsdam.swp_webspeiseplan_parser import SWPWebspeiseplanParser
|
||||||
from stw_potsdam.config import Canteen
|
from stw_potsdam.config import Canteen
|
||||||
from stw_potsdam.xml_types.feed_xml import FeedXML, ScheduleXML
|
from stw_potsdam.xml_types.feed_xml import FeedXML, ScheduleXML
|
||||||
@@ -15,21 +18,26 @@ class Builder:
|
|||||||
|
|
||||||
VERSION = "2.0.1"
|
VERSION = "2.0.1"
|
||||||
|
|
||||||
def __init__(self, config: dict[str, Canteen]):
|
def __init__(
|
||||||
|
self,
|
||||||
|
config: dict[str, Canteen],
|
||||||
|
swp_data: SWPWebspeiseplanData | None = None,
|
||||||
|
):
|
||||||
"""Initialize the object for the OpenMensa Feed Doc XML."""
|
"""Initialize the object for the OpenMensa Feed Doc XML."""
|
||||||
logging.basicConfig()
|
logging.basicConfig()
|
||||||
self.logger = logging.getLogger(__name__)
|
self.logger = logging.getLogger(__name__)
|
||||||
self._xml_data = {}
|
self._xml_data = {}
|
||||||
swp_api = SWPWebspeiseplanAPI()
|
if swp_data is None:
|
||||||
|
swp_data = SWPWebspeiseplanAPI().fetch_all()
|
||||||
swp_parser = SWPWebspeiseplanParser()
|
swp_parser = SWPWebspeiseplanParser()
|
||||||
for cname, ntup in config.items():
|
for cname, ntup in config.items():
|
||||||
if ntup.name not in swp_api.outlets.keys():
|
if ntup.name not in swp_data.outlets.keys():
|
||||||
self.logger.warning("%s not found in keys", ntup.name)
|
self.logger.warning("%s not found in keys", ntup.name)
|
||||||
continue
|
continue
|
||||||
outlet = swp_api.outlets[ntup.name]
|
outlet = swp_data.outlets[ntup.name]
|
||||||
menus = swp_api.menus[ntup.name]
|
menus = swp_data.menus[ntup.name]
|
||||||
categories = swp_api.meal_categories[ntup.name]
|
categories = swp_data.meal_categories[ntup.name]
|
||||||
locations = swp_api.locations[ntup.name]
|
locations = swp_data.locations[ntup.name]
|
||||||
outlet["isPublic"] = locations["isPublic"]
|
outlet["isPublic"] = locations["isPublic"]
|
||||||
canteen = swp_parser.parse_canteen_meta_times(outlet)
|
canteen = swp_parser.parse_canteen_meta_times(outlet)
|
||||||
meals = swp_parser.parse_meals(menus, categories)
|
meals = swp_parser.parse_meals(menus, categories)
|
||||||
|
|||||||
@@ -0,0 +1,12 @@
|
|||||||
|
# -*- encoding: utf-8 -*-
|
||||||
|
|
||||||
|
from stw_potsdam.swp_webspeiseplan_api import SWPWebspeiseplanAPI
|
||||||
|
|
||||||
|
# pytest fixtures are linked via parameter names of test methods
|
||||||
|
# pragma pylint: disable=unused-import,unused-argument,redefined-outer-name
|
||||||
|
from tests.stub_api import api_offline
|
||||||
|
|
||||||
|
|
||||||
|
def test_api_init_does_not_fetch(api_offline):
|
||||||
|
"""Creating the API client does not perform network requests."""
|
||||||
|
SWPWebspeiseplanAPI()
|
||||||
Reference in New Issue
Block a user