piket/piket_client/model.py

"""
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


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