piket/piket_server/models.py

"""
Defines database models used by the server.
"""

import datetime
from typing import List, Dict, Any
from collections import defaultdict

from sqlalchemy import func
from sqlalchemy.exc import SQLAlchemyError

from piket_server.flask import db


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

    __tablename__ = "people"

    person_id = db.Column(db.Integer, primary_key=True)
    full_name = db.Column(db.String, nullable=False)
    display_name = db.Column(db.String, nullable=True)
    aardbei_id = db.Column(db.Integer, nullable=True)
    active = db.Column(db.Boolean, nullable=False, default=False)

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

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

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


class Export(db.Model):
    """ Represents a set of exported Settlements. """

    __tablename__ = "exports"

    export_id = db.Column(db.Integer, primary_key=True)
    created_at = db.Column(
        db.DateTime, default=datetime.datetime.utcnow, nullable=False
    )

    settlements = db.relationship("Settlement", backref="export", lazy=True)

    @property
    def as_dict(self) -> dict:
        return {
            "export_id": self.export_id,
            "created_at": self.created_at.isoformat(),
            "settlement_ids": [s.settlement_id for s in self.settlements],
        }


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)
    export_id = db.Column(db.Integer, db.ForeignKey("exports.export_id"), nullable=True)

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

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

    @property
    def as_dict(self) -> dict:
        return {
            "settlement_id": self.settlement_id,
            "name": self.name,
            "consumption_summary": self.consumption_summary,
            "unique_people": self.unique_people,
            "per_person_counts": self.per_person_counts,
            "count_info": self.per_person,
        }

    @property
    def unique_people(self) -> int:
        q = (
            Consumption.query.filter_by(settlement=self)
            .filter_by(reversed=False)
            .group_by(Consumption.person_id)
            .count()
        )
        return q

    @property
    def consumption_summary(self) -> dict:
        q = (
            Consumption.query.filter_by(settlement=self)
            .filter_by(reversed=False)
            .group_by(Consumption.consumption_type_id)
            .order_by(ConsumptionType.name)
            .outerjoin(ConsumptionType)
            .with_entities(
                Consumption.consumption_type_id,
                ConsumptionType.name,
                func.count(Consumption.consumption_id),
            )
            .all()
        )

        return {r[0]: {"name": r[1], "count": r[2]} for r in q}

    @property
    def per_person(self) -> dict:
        # Get keys of seen consumption_types
        c_types = self.consumption_summary.keys()

        result = {}
        for type in c_types:
            c_type = ConsumptionType.query.get(type)
            result[type] = {"consumption_type": c_type.as_dict, "counts": {}}

            q = (
                Consumption.query.filter_by(settlement=self)
                .filter_by(reversed=False)
                .filter_by(consumption_type=c_type)
                .group_by(Consumption.person_id)
                .order_by(Person.full_name)
                .outerjoin(Person)
                .with_entities(
                    Person.person_id,
                    Person.full_name,
                    func.count(Consumption.consumption_id),
                )
                .all()
            )

            for row in q:
                result[type]["counts"][row[0]] = {"name": row[1], "count": row[2]}

        return result

    @property
    def per_person_counts(self) -> Dict[int, Any]:
        """
        Output a more usable dict containing for each person in the settlement
        how many of each consumption type was counted.
        """

        q = (
            Consumption.query.filter_by(settlement=self)
            .filter_by(reversed=False)
            .group_by(Consumption.person_id)
            .group_by(Consumption.consumption_type_id)
            .group_by(Person.full_name)
            .outerjoin(Person)
            .with_entities(
                Consumption.person_id,
                Person.full_name,
                Consumption.consumption_type_id,
                func.count(Consumption.consumption_id),
            )
            .all()
        )

        res: Dict[int, Any] = defaultdict(dict)

        for row in q:
            item = res[row[0]]
            item["full_name"] = row[1]
            if not item.get("counts"):
                item["counts"] = {}

            item["counts"][row[2]] = row[3]

        return res




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)
    active = db.Column(db.Boolean, default=True)

    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,
            "active": self.active,
        }


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.full_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,
        }