vacation-planner/server/app.py

import os
import datetime
import itertools
from typing import List, Union
from enum import Enum

from fastapi import FastAPI, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import aiosqlite

DB_FILE = os.environ.get("DB_LOCATION", "vacation.db")
LAST_UPDATE = datetime.datetime.now()

app = FastAPI()
app.add_middleware(
    CORSMiddleware,     
    allow_origins=["*"],
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)


def connect():
    return aiosqlite.connect(DB_FILE)


@app.on_event("startup")
async def setup():
    async with connect() as db:
        await db.execute("CREATE TABLE IF NOT EXISTS availability ( \
            name TEXT NOT NULL, \
            month INTEGER NOT NULL, \
            day INTEGER NOT NULL, \
            status TEXT CHECK(status = 'yes' OR status = 'no' OR status = 'maybe' OR status = 'unknown') DEFAULT 'unknown', \
            PRIMARY KEY (name, month, day)\
        );")
        await db.commit()


@app.get("/")
def health():
    return {"status": "healthy"}


@app.get("/availability")
async def get_all():
    results = []
    async with connect() as db:
        async with db.execute("SELECT name, month, day, status FROM availability ORDER BY month, day;") as cursor:
            results = await cursor.fetchall()
    return {
        "lastUpdated": LAST_UPDATE.isoformat(),
        "availability": [
            {
                "month": month,
                "day": day,
                "availability": [{
                    "name": name,
                    "status": status,
                } for (name, _, _, status) in avail],
            } for ((month, day), avail) in itertools.groupby(results, key=lambda t: (t[1], t[2]))
        ],
    }


class AvailabilityStatus(Enum):
    yes = "yes"
    maybe = "maybe"
    no = "no"
    unknown = "unknown"


class AvailabilityItem(BaseModel):
    month: int
    day: int
    status: Union[AvailabilityStatus, None]


class Availability(BaseModel):
    name: str
    availability: List[AvailabilityItem]


@app.post("/availability", status_code=status.HTTP_204_NO_CONTENT)
async def set_availability(body: Availability):
    global LAST_UPDATE
    async with connect() as db:
        for a in body.availability:
            await db.execute(
                "INSERT INTO availability(name, month, day, status) VALUES (?, ?, ?, ?) ON CONFLICT DO UPDATE SET name=?, month=?, day=?, status=?", 
                (body.name, a.month, a.day, a.status.value, body.name, a.month, a.day, a.status.value),
            )
        await db.commit()
    LAST_UPDATE = datetime.datetime.now()