''' 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') )