61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from src.components.poo_recorder import PooRecorder
|
|
from src.config import Config
|
|
from src.util.homeassistant import HomeAssistant
|
|
from src.util.location_recorder import LocationRecorder
|
|
from src.util.mqtt import MQTT
|
|
from src.util.notion import NotionAsync
|
|
from src.util.ticktick import TickTick
|
|
|
|
Config.init()
|
|
|
|
location_recorder_db = str(Path(__file__).resolve().parent / ".." / "location_recorder.db")
|
|
|
|
ticktick = TickTick()
|
|
notion = NotionAsync(token=Config.get_env(key="NOTION_TOKEN"))
|
|
mqtt = MQTT()
|
|
location_recorder = LocationRecorder(db_path=location_recorder_db)
|
|
homeassistant = HomeAssistant(ticktick=ticktick, location_recorder=location_recorder)
|
|
poo_recorder = PooRecorder(mqtt=mqtt, notion=notion, homeassistant=homeassistant)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _lifespan(_app: FastAPI): # noqa: ANN202
|
|
await mqtt.start()
|
|
await location_recorder.create_db_engine()
|
|
yield
|
|
await mqtt.stop()
|
|
await location_recorder.dispose_db_engine()
|
|
|
|
|
|
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.Message) -> dict:
|
|
return await homeassistant.process_message(message=payload)
|
|
|
|
|
|
# Poo recorder
|
|
@app.post("/poo/record")
|
|
async def record(record_detail: PooRecorder.RecordField) -> PooRecorder.RecordField:
|
|
await poo_recorder.record(record_detail)
|
|
return record_detail
|
|
|
|
|
|
# ticktick
|
|
@app.get("/ticktick/auth/code")
|
|
async def ticktick_auth(code: str, state: str) -> dict:
|
|
if await ticktick.retrieve_access_token(code, state):
|
|
return {"State": "Token Retrieved"}
|
|
return {"State": "Token Retrieval Failed"}
|