59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
from pydantic import BaseModel
|
|
|
|
from src.cloud_util.homeassistant import HomeAssistant
|
|
from src.cloud_util.mqtt import MQTT
|
|
from src.cloud_util.notion import NotionAsync
|
|
from src.cloud_util.ticktick import TickTick
|
|
from src.config import Config
|
|
from src.recorder.poo import PooRecorder
|
|
|
|
Config.init()
|
|
|
|
ticktick = TickTick()
|
|
notion = NotionAsync(token=Config.get_env(key="NOTION_TOKEN"))
|
|
mqtt = MQTT()
|
|
poo_recorder = PooRecorder(mqtt=mqtt, notion=notion)
|
|
homeassistant = HomeAssistant(ticktick=ticktick)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _lifespan(_app: FastAPI): # noqa: ANN202
|
|
await mqtt.start()
|
|
yield
|
|
await mqtt.stop()
|
|
|
|
|
|
class PooRecordField(BaseModel):
|
|
status: str
|
|
|
|
|
|
app = FastAPI(lifespan=_lifespan)
|
|
|
|
|
|
@app.get("/homeassistant/status")
|
|
async def get_status() -> dict:
|
|
return {"Status": "Ok"}
|
|
|
|
|
|
@app.post("/homeassistant/publish")
|
|
async def homeassistant_publish(payload: HomeAssistant.PublishMessage) -> dict:
|
|
return homeassistant.process_publish_message(message=payload)
|
|
|
|
|
|
# Poo recorder
|
|
@app.post("/poo/record")
|
|
async def record(record_detail: PooRecordField) -> PooRecordField:
|
|
await poo_recorder.record(record_detail.status)
|
|
return record_detail
|
|
|
|
|
|
# ticktick
|
|
@app.get("/ticktick/auth/code")
|
|
async def ticktick_auth(code: str, state: str) -> dict:
|
|
if ticktick.retrieve_access_token(code, state):
|
|
return {"State": "Token Retrieved"}
|
|
return {"State": "Token Retrieval Failed"}
|