""" Provides access to the models stored in the database, via the server. """ import datetime import logging from typing import NamedTuple from urllib.parse import urljoin import requests LOG = logging.getLogger(__name__) SERVER_URL = "http://127.0.0.1:5000" class ServerStatus: """ Provides helper classes to check whether the server is up. """ @classmethod def is_server_running(cls) -> bool: try: req = requests.get(urljoin(SERVER_URL, "ping")) if req.status_code == 200: return True, req.content return False, req.content except requests.ConnectionError as ex: return False, ex datetime_format = "%Y-%m-%dT%H:%M:%S.%f" @classmethod def unsettled_consumptions(cls) -> dict: req = requests.get(urljoin(SERVER_URL, 'status')) data = req.json() if data['unsettled']['amount']: data['unsettled']['first'] = datetime.datetime\ .strptime(data['unsettled']['first'], cls.datetime_format) data['unsettled']['last'] = datetime.datetime\ .strptime(data['unsettled']['last'], cls.datetime_format) return data 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 class Settlement(NamedTuple): """ Represents a stored Settlement. """ settlement_id: int name: str consumption_summary: dict @classmethod def from_dict(cls, data: dict) -> "Settlement": return Settlement( settlement_id=data['settlement_id'], name=data['name'], consumption_summary=data['consumption_summary'] ) @classmethod def create(cls, name: str) -> "Settlement": req = requests.post( urljoin(SERVER_URL, '/settlements'), json={'settlement': {'name': name}} ) return cls.from_dict(req.json()['settlement'])