tcl-filtrage/main.py

import enum
import re
from typing import Any, DefaultDict, List, Optional

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.params import Header, Path
from pydantic import BaseModel

app = FastAPI()


class Passage(BaseModel):
    ligne: str
    delais: List[str]


class Passages(BaseModel):
    passages: List[Passage]
    stop_id: int


@app.get("/stop/{stop_id}", response_model=Passages)
async def stop(
    stop_id: int = Path(
        None,
        description="Stop id to monitor. Can be obtained using https://data.grandlyon.com/jeux-de-donnees/points-arret-reseau-transports-commun-lyonnais/donnees",
    ),
    authorization: Optional[str] = Header(
        None,
        alias="Authorization",
        description="Basic auth for remote API (data grand lyon)",
    ),
):
    if authorization is None:
        raise HTTPException(status_code=401, detail="Not authenticated")

    headers = {"Authorization": authorization}
    async with httpx.AsyncClient(headers=headers) as client:
        res = await client.get(
            "https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclpassagearret/all.json?maxfeatures=-1"
        )
        if res.status_code != 200:
            raise HTTPException(
                status_code=res.status_code,
                detail="HTTP error during call to remote API",
            )

    passages = DefaultDict(list)
    for passage in res.json().get("values"):
        if passage.get("id") == stop_id:
            ligne = passage.get("ligne")
            ligne = re.sub(
                "[A-Z]$", "", ligne
            )  # Remove letter suffix to group by commercial line name
            passages[ligne].append(passage.get("delaipassage"))
    passages_list = []
    for ligne, delais in passages.items():
        passages_list.append(Passage(ligne=ligne, delais=delais))
    return Passages(passages=passages_list, stop_id=stop_id)