""" 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, unique=True) display_name = db.Column(db.String, nullable=True, unique=True) aardbei_id = db.Column(db.Integer, nullable=True, unique=True) active = db.Column(db.Boolean, nullable=False, default=False) consumptions = db.relationship("Consumption", backref="person", lazy=True) def __repr__(self) -> str: return f"" @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"" @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, unique=True) 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"" @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"" ) @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, }