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

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

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


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


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


PRESET_NAMES = [
    "Maarten",
    "Knoepie Draggelsturf",
    "Teddy Veenlijk",
    "Chris Kraslot",
    "Knibbe Tjakkomans",
    "Foek Lammenschaap",
]

PEOPLE = {
    index: {"id": index, "name": name, "count": 0}
    for index, name in enumerate(PRESET_NAMES)
}
NEXT_ID = len(PEOPLE)


@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."})

    return jsonify(person=person.as_dict)


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