""" Provides access to the models stored in the database, via the server. """ from typing import NamedTuple from urllib.parse import urljoin import requests from . import logger LOG = logger.getLogger("model") SERVER_URL = "http://127.0.0.1:5000" class Person(NamedTuple): """ Represents a Person, as retrieved from the database. """ name: str person_id: int = None consumptions: dict = {} def add_consumption(self, type_id: str) -> bool: """ Register a consumption for this Person. """ req = requests.post( urljoin(SERVER_URL, f"people/{self.person_id}/add_consumption/{type_id}") ) try: data = req.json() if "error" in data: LOG.error( "Could not add consumption for %s (%s): %s", self.person_id, req.status_code, data, ) return False self.consumptions.update(data["person"]["consumptions"]) return True except ValueError: LOG.error( "Did not get JSON on adding Consumption (%s): %s", req.status_code, req.content, ) return False def create(self) -> "Person": """ Create a new Person from the current attributes. As tuples are immutable, a new Person with the correct id is returned. """ req = requests.post( urljoin(SERVER_URL, "people"), json={"person": {"name": self.name}} ) try: data = req.json() except ValueError: LOG.error( "Did not get JSON on adding Person (%s): %s", req.status_code, req.content, ) return None if "error" in data or req.status_code != 201: LOG.error("Could not create Person (%s): %s", req.status_code, data) return None return Person.from_dict(data["person"]) @classmethod def get(cls, person_id: int) -> "Person": """ Retrieve a Person by id. """ req = requests.get(urljoin(SERVER_URL, f"/people/{person_id}")) try: data = req.json() if "error" in data: LOG.warning( "Could not get person %s (%s): %s", person_id, req.status_code, data ) return None return Person.from_dict(data["person"]) except ValueError: LOG.error( "Did not get JSON from server on getting Person (%s): %s", req.status_code, req.content, ) return None @classmethod def get_all(cls) -> ["Person"]: """ Get all active People. """ req = requests.get(urljoin(SERVER_URL, "/people")) try: data = req.json() if "error" in data: LOG.warning("Could not get people (%s): %s", req.status_code, data) return [Person.from_dict(item) for item in data["people"]] except ValueError: LOG.error( "Did not get JSON from server on getting People (%s): %s", req.status_code, req.content, ) return None @classmethod def from_dict(cls, data: dict) -> "Person": """ Reconstruct a Person object from a dict. """ return Person( name=data["name"], person_id=data["person_id"], consumptions=data["consumptions"], ) class ConsumptionType(NamedTuple): """ Represents a stored ConsumptionType. """ name: str consumption_type_id: int = None icon: str = None def create(self) -> "Person": """ Create a new ConsumptionType from the current attributes. As tuples are immutable, a new ConsumptionType with the correct id is returned. """ req = requests.post( urljoin(SERVER_URL, "consumption_types"), json={"consumption_type": {"name": self.name, "icon": self.icon}}, ) try: data = req.json() except ValueError: LOG.error( "Did not get JSON on adding ConsumptionType (%s): %s", req.status_code, req.content, ) return None if "error" in data or req.status_code != 201: LOG.error( "Could not create ConsumptionType (%s): %s", req.status_code, data ) return None return Person.from_dict(data["consumption_type"]) @classmethod def get(cls, consumption_type_id: int) -> "ConsumptionType": """ Retrieve a ConsumptionType by id. """ req = requests.get( urljoin(SERVER_URL, f"/consumption_types/{consumption_type_id}") ) try: data = req.json() if "error" in data: LOG.warning( "Could not get consumption type %s (%s): %s", consumption_type_id, req.status_code, data, ) return None return cls.from_dict(data["consumption_type"]) except ValueError: LOG.error( "Did not get JSON from server on getting consumption type (%s): %s", req.status_code, req.content, ) return None @classmethod def get_all(cls) -> ["ConsumptionType"]: """ Get all active ConsumptionTypes. """ req = requests.get(urljoin(SERVER_URL, "/consumption_types")) try: data = req.json() if "error" in data: LOG.warning( "Could not get consumption types (%s): %s", req.status_code, data ) return [cls.from_dict(item) for item in data["consumption_types"]] except ValueError: LOG.error( "Did not get JSON from server on getting ConsumptionTypes (%s): %s", req.status_code, req.content, ) return None @classmethod def from_dict(cls, data: dict) -> "ConsumptionType": """ Reconstruct a ConsumptionType from a dict. """ return ConsumptionType( name=data["name"], consumption_type_id=data["consumption_type_id"], icon=data.get("icon"), )