piket/piket_client/model.py

"""
Provides access to the models stored in the database, via the server.
"""
from __future__ import annotations

import datetime
import logging
from typing import Any, List, NamedTuple, Optional, Sequence, Tuple
from urllib.parse import urljoin

import requests

LOG = logging.getLogger(__name__)

SERVER_URL = "http://127.0.0.1:5000"
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"


class ServerStatus:
    """ Provides helper classes to check whether the server is up. """

    @classmethod
    def is_server_running(cls) -> Tuple[bool, Any]:
        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

    @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"], DATETIME_FORMAT
            )
            data["unsettled"]["last"] = datetime.datetime.strptime(
                data["unsettled"]["last"], DATETIME_FORMAT
            )

        return data


class Person(NamedTuple):
    """ Represents a Person, as retrieved from the database. """

    full_name: str
    display_name: Optional[str]
    active: bool = True
    person_id: Optional[int] = None
    consumptions: dict = {}

    @property
    def name(self) -> str:
        return self.display_name or self.full_name

    def add_consumption(self, type_id: str) -> Optional[Consumption]:
        """ 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 None

            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 None

    def create(self) -> Optional[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, "active": True}},
        )

        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) -> Optional[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) -> Optional[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) -> Optional[List[Person]]:
        """ Get all active People. """
        params = {}
        if active is not None:
            params["active"] = int(active)

        req = requests.get(urljoin(SERVER_URL, "/people"), params=params)

        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(
            full_name=data["full_name"],
            display_name=data["display_name"],
            active=data["active"],
            person_id=data["person_id"],
            consumptions=data["consumptions"],
        )


class Export(NamedTuple):
    created_at: datetime.datetime
    settlement_ids: Sequence[int]
    export_id: int
    settlements: Sequence["Settlement"] = []

    @classmethod
    def from_dict(cls, data: dict) -> "Export":
        """ Reconstruct an Export from a dict. """
        return cls(
            export_id=data["export_id"],
            created_at=datetime.datetime.strptime(data["created_at"], DATETIME_FORMAT),
            settlement_ids=data["settlement_ids"],
            settlements=data.get("settlements", []),
        )

    @classmethod
    def get_all(cls) -> Optional[List[Export]]:
        """ Get a list of all existing Exports. """
        req = requests.get(urljoin(SERVER_URL, "exports"))

        try:
            data = req.json()
        except ValueError:
            LOG.error(
                "Did not get JSON on listing Exports (%s): %s",
                req.status_code,
                req.content,
            )
            return None

        if "error" in data or req.status_code != 200:
            LOG.error("Could not list Exports (%s): %s", req.status_code, data)
            return None

        return [cls.from_dict(e) for e in data["exports"]]

    @classmethod
    def get(cls, export_id: int) -> Optional[Export]:
        """ Retrieve one Export. """
        req = requests.get(urljoin(SERVER_URL, f"exports/{export_id}"))

        try:
            data = req.json()
        except ValueError:
            LOG.error(
                "Did not get JSON on getting Export (%s): %s",
                req.status_code,
                req.content,
            )
            return None

        if "error" in data or req.status_code != 200:
            LOG.error("Could not get Export (%s): %s", req.status_code, data)
            return None

        data["export"]["settlements"] = data["settlements"]

        return cls.from_dict(data["export"])

    @classmethod
    def create(cls) -> Optional[Export]:
        """ Create a new Export, containing all un-exported Settlements. """
        req = requests.post(urljoin(SERVER_URL, "exports"))

        try:
            data = req.json()
        except ValueError:
            LOG.error(
                "Did not get JSON on adding Export (%s): %s",
                req.status_code,
                req.content,
            )
            return None

        if "error" in data or req.status_code != 201:
            LOG.error("Could not create Export (%s): %s", req.status_code, data)
            return None

        data["export"]["settlements"] = data["settlements"]

        return cls.from_dict(data["export"])


class ConsumptionType(NamedTuple):
    """ Represents a stored ConsumptionType. """

    name: str
    consumption_type_id: Optional[int] = None
    icon: Optional[str] = None

    def create(self) -> Optional[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) -> Optional[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) -> Optional[List[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: Optional[int] = None

    @classmethod
    def from_dict(cls, data: dict) -> "Consumption":
        """ Reconstruct a Consumption from a dict. """
        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) -> Optional[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 None

            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 None


class Settlement(NamedTuple):
    """ Represents a stored Settlement. """

    settlement_id: int
    name: str
    consumption_summary: dict
    count_info: dict = {}

    @classmethod
    def from_dict(cls, data: dict) -> "Settlement":
        return Settlement(
            settlement_id=data["settlement_id"],
            name=data["name"],
            consumption_summary=data["consumption_summary"],
            count_info=data.get("count_info", {}),
        )

    @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"])

    @classmethod
    def get(cls, settlement_id: int) -> Optional[Settlement]:
        req = requests.get(urljoin(SERVER_URL, f"/settlements/{settlement_id}"))

        try:
            data = req.json()
        except ValueError:
            LOG.error(
                "Did not get JSON on retrieving Settlement (%s): %s",
                req.status_code,
                req.content,
            )
            return None

        if "error" in data or req.status_code != 200:
            LOG.error("Could not get Export (%s): %s", req.status_code, data)
            return None

        data["settlement"]["count_info"] = data["count_info"]

        return cls.from_dict(data["settlement"])