piket/piket_server/__init__.py

"""
Piket server, handles events generated by the client.
"""

import datetime
import os

from sqlalchemy.exc import SQLAlchemyError
from flask import Flask, jsonify, abort, request
from flask_sqlalchemy import SQLAlchemy


DATA_HOME = os.environ.get("XDG_DATA_HOME", "~/.local/share")
CONFIG_DIR = os.path.join(DATA_HOME, "piket_server")
DB_PATH = os.path.expanduser(os.path.join(CONFIG_DIR, "database.sqlite3"))
DB_URL = f"sqlite:///{DB_PATH}"

app = Flask("piket_server")
app.config["SQLALCHEMY_DATABASE_URI"] = DB_URL
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)


# ---------- Models ----------
class Person(db.Model):
    """ Represents a person to be shown on the lists. """

    __tablename__ = "people"

    person_id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)

    consumptions = db.relationship("Consumption", backref="person", lazy=True)

    def __repr__(self) -> str:
        return f"<Person {self.person_id}: {self.name}>"

    @property
    def as_dict(self) -> dict:
        return {
            "person_id": self.person_id,
            "name": self.name,
            "consumptions": {
                ct.consumption_type_id: Consumption.query.filter_by(person=self)
                .filter_by(consumption_type=ct)
                .filter_by(reversed=False)
                .count()
                for ct in ConsumptionType.query.all()
            },
        }


class Settlement(db.Model):
    """ Represents a settlement of the list. """

    __tablename__ = "settlements"

    settlement_id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)

    consumptions = db.relationship("Consumption", backref="settlement", lazy=True)

    def __repr__(self) -> str:
        return f"<Settlement {self.settlement_id}: {self.name}>"


class ConsumptionType(db.Model):
    """ Represents a type of consumption to be counted. """

    __tablename__ = "consumption_types"

    consumption_type_id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)
    icon = db.Column(db.String)

    consumptions = db.relationship("Consumption", backref="consumption_type", lazy=True)

    def __repr__(self) -> str:
        return f"<ConsumptionType: {self.name}>"

    @property
    def as_dict(self) -> dict:
        return {
            "consumption_type_id": self.consumption_type_id,
            "name": self.name,
            "icon": self.icon,
        }


class Consumption(db.Model):
    """ Represent one consumption to be counted. """

    __tablename__ = "consumptions"

    consumption_id = db.Column(db.Integer, primary_key=True)
    person_id = db.Column(db.Integer, db.ForeignKey("people.person_id"), nullable=True)
    consumption_type_id = db.Column(
        db.Integer,
        db.ForeignKey("consumption_types.consumption_type_id"),
        nullable=False,
    )
    settlement_id = db.Column(
        db.Integer, db.ForeignKey("settlements.settlement_id"), nullable=True
    )
    created_at = db.Column(
        db.DateTime, default=datetime.datetime.utcnow, nullable=False
    )
    reversed = db.Column(db.Boolean, default=False, nullable=False)

    def __repr__(self) -> str:
        return f"<Consumption: {self.consumption_type.name} for {self.person.name}>"

    @property
    def as_dict(self) -> dict:
        return {
            "consumption_id": self.consumption_id,
            "person_id": self.person_id,
            "consumption_type_id": self.consumption_type_id,
            "settlement_id": self.settlement_id,
            "created_at": self.created_at.isoformat(),
            "reversed": self.reversed,
        }


# ---------- Models ----------


@app.route("/ping")
def ping() -> None:
    """ Return a status ping. """
    return "Pong"


# Person
@app.route("/people", methods=["GET"])
def get_people():
    """ Return a list of currently known people. """
    people = Person.query.order_by(Person.name).all()
    result = [person.as_dict for person in people]
    return jsonify(people=result)


@app.route("/people/<int:person_id>", methods=["GET"])
def get_person(person_id: int):
    person = Person.query.get_or_404(person_id)

    return jsonify(person=person.as_dict)


@app.route("/people", methods=["POST"])
def add_person():
    """
    Add a new person.

    Required parameters:
    - name (str)
    """
    json = request.get_json()

    if not json:
        return jsonify({"error": "Could not parse JSON."}), 400

    data = json.get("person") or {}
    person = Person(name=data.get("name"))
    try:
        db.session.add(person)
        db.session.commit()
    except SQLAlchemyError:
        return jsonify({"error": "Invalid arguments for Person."}), 400

    return jsonify(person=person.as_dict), 201


@app.route("/people/<int:person_id>/add_consumption", methods=["POST"])
def add_consumption(person_id: int):
    person = Person.query.get_or_404(person_id)

    consumption = Consumption(person=person, consumption_type_id=1)
    try:
        db.session.add(consumption)
        db.session.commit()
    except SQLAlchemyError:
        return (
            jsonify(
                {"error": "Invalid Consumption parameters.", "person": person.as_dict}
            ),
            400,
        )

    return jsonify(person=person.as_dict, consumption=consumption.as_dict), 201


@app.route("/people/<int:person_id>/add_consumption/<int:ct_id>", methods=["POST"])
def add_consumption2(person_id: int, ct_id: int):
    person = Person.query.get_or_404(person_id)

    consumption = Consumption(person=person, consumption_type_id=ct_id)
    try:
        db.session.add(consumption)
        db.session.commit()
    except SQLAlchemyError:
        return (
            jsonify(
                {"error": "Invalid Consumption parameters.", "person": person.as_dict}
            ),
            400,
        )

    return jsonify(person=person.as_dict, consumption=consumption.as_dict), 201


@app.route("/consumptions/<int:consumption_id>", methods=["DELETE"])
def reverse_consumption(consumption_id: int):
    """ Reverse a consumption. """
    consumption = Consumption.query.get_or_404(consumption_id)

    if consumption.reversed:
        return (
            jsonify(
                {
                    "error": "Consumption already reversed",
                    "consumption": consumption.as_dict,
                }
            ),
            409,
        )

    try:
        consumption.reversed = True
        db.session.add(consumption)
        db.session.commit()

    except SQLAlchemyError:
        return jsonify({"error": "Database error."}), 500

    return jsonify(consumption=consumption.as_dict), 200


# ConsumptionType
@app.route("/consumption_types", methods=["GET"])
def get_consumption_types():
    """ Return a list of currently active consumption types. """
    ctypes = ConsumptionType.query.all()
    result = [ct.as_dict for ct in ctypes]
    return jsonify(consumption_types=result)


@app.route("/consumption_types/<int:consumption_type_id>", methods=["GET"])
def get_consumption_type(consumption_type_id: int):
    ct = ConsumptionType.query.get_or_404(consumption_type_id)

    return jsonify(consumption_type=ct.as_dict)


@app.route("/consumption_types", methods=["POST"])
def add_consumption_type():
    """ Add a new ConsumptionType.  """
    json = request.get_json()

    if not json:
        return jsonify({"error": "Could not parse JSON."}), 400

    data = json.get("consumption_type") or {}
    ct = ConsumptionType(name=data.get("name"), icon=data.get("icon"))

    try:
        db.session.add(ct)
        db.session.commit()
    except SQLAlchemyError:
        return jsonify({"error": "Invalid arguments for ConsumptionType."}), 400

    return jsonify(consumption_type=ct.as_dict), 201