iac_test/parser_api/router.py

75 lines
1.9 KiB
Python

from fastapi import HTTPException, Depends
from sqlalchemy.orm import Session
from typing import List, Annotated
from fastapi import APIRouter
from . import models, schemas, controller
from .database import SessionLocal, 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)