from typing import Annotated, List from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from . import controller, schemas from .database import get_db router = APIRouter(prefix="/api") @router.get("/list", response_model=List[schemas.Record], summary="Search by filters") def list_rows( filters: Annotated[schemas.RecordRequest, Depends()], db: Session = Depends(get_db) ): """ Searches rows with specified filters. Case insensitive contains: - **region** - **area** - **town** - **street** - **start** - **finish** - **branch** - **res** - **comment** Exact match: - **index** - **building_id** - **lat** - **lng** Later or earlier than respectively: - **start** - **finish** """ return controller.search_each(db, filters) @router.get("/search", response_model=List[schemas.Record], summary="Search by query") def search_rows(query: str, db: Session = Depends(get_db)): """ Selects rows with cells containing case insensitive prompt as its part. In addition, geocoding is being applied to prompt and if building_id found, corresponding row is being returned. Rows to be searched: - **region** - **area** - **town** - **street** - **branch** - **res** - **comment** """ return controller.search_all(db, query) @router.get( "/check", response_model=schemas.CheckResponse, summary="Check when outage ends" ) def check(building_id: int, db: Session = Depends(get_db)): """ Checks if there is an active outage for building_id and if there is, also returns when will it end """ return controller.check_outage(db, building_id) @router.put("/create", response_model=schemas.Record) def create_record(record: schemas.RecordCreate, db: Session = Depends(get_db)): """ Not for public usage """ return controller.create_record(db, record)