""" Provides access to the models stored in the database, via the server. """ import datetime 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 Consumption.from_dict(data["consumption"]) 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"]) def set_active(self, new_state=True) -> "Person": req = requests.patch(urljoin(SERVER_URL, f"people/{self.person_id}"), json={'person': {'active':new_state}}) try: data = req.json() except ValueError: LOG.error( "Did not get JSON on updating Person (%s): %s", req.status_code, req.content, ) return None if "error" in data or req.status_code != 200: LOG.error("Could not update 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, active=None) -> ["Person"]: """ Get all active People. """ active = int(active) req = requests.get(urljoin(SERVER_URL, "/people"), params={'active':active}) 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) -> "ConsumptionType": """ 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 ConsumptionType.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 cls( name=data["name"], consumption_type_id=data["consumption_type_id"], icon=data.get("icon"), ) class Consumption(NamedTuple): """ Represents a stored Consumption. """ consumption_id: int person_id: int consumption_type_id: int created_at: datetime.datetime reversed: bool = False settlement_id: int = None @classmethod def from_dict(cls, data: dict) -> "Consumption": """ Reconstruct a Consumption from a dict. """ datetime_format = '%Y-%m-%dT%H:%M:%S.%f' # 2018-08-31T17:30:47.871521 return cls( consumption_id=data["consumption_id"], person_id=data["person_id"], consumption_type_id=data["consumption_type_id"], settlement_id=data["settlement_id"], created_at=datetime.datetime.strptime(data["created_at"], datetime_format), reversed=data["reversed"], ) def reverse(self) -> "Consumption": """ Reverse this consumption. """ req = requests.delete( urljoin(SERVER_URL, f"/consumptions/{self.consumption_id}") ) try: data = req.json() if "error" in data: LOG.error( "Could not reverse consumption %s (%s): %s", self.consumption_id, req.status_code, data, ) return False return Consumption.from_dict(data["consumption"]) except ValueError: LOG.error( "Did not get JSON on reversing Consumption (%s): %s", req.status_code, req.content, ) return False