Added routes descriptions and healthcheck schema

parser_api/controller.py

@@ -8,6 +8,8 @@ from sqlalchemy.orm import Session
 from sqlalchemy.sql import operators
 from sqlalchemy.sql.expression import BinaryExpression
 
+from rosseti_parser import get_building_id
+
 from . import models, schemas
 
 
@@ -40,7 +42,7 @@ def contains_lower(name, val):
         return getattr(models.Record, name) == val
 
 
-def and_if_can(a: BinaryExpression, b: Optional[BinaryExpression]):
+def and_if_can(a: BinaryExpression, b: Optional[BinaryExpression]) -> BinaryExpression:
     if b is not None:
         return a & b
     else:
@@ -63,8 +65,8 @@ def search_each(db: Session, filters: schemas.RecordRequest) -> List[schemas.Rec
 
     if query is None:
         res = db.query(models.Record).all()
-
-    res = db.query(models.Record).filter(query).all()
+    else:
+        res = db.query(models.Record).filter(query).all()
 
     return res
 
@@ -81,6 +83,11 @@ def search_all(db: Session, prompt: str) -> List[schemas.Record]:
         'res'
     ), contains_lower('comment', prompt))
 
+    building_id, *_ = get_building_id(prompt)
+
+    if building_id is not None:
+        query |= models.Record.building_id == building_id
+
     res = db.query(models.Record).filter(query).all()
 
     return res

parser_api/main.py

@@ -4,7 +4,7 @@ import datetime
 from fastapi import FastAPI
 import schedule
 
-from . import models, router
+from . import models, router, schemas
 from .database import engine
 from .scheduler import run_continuously, run_threaded
 from .job import job
@@ -30,8 +30,8 @@ app = FastAPI(lifespan=lifespan)
 app.include_router(router.router)
 
 
-@app.get('/')
-def root():
+@app.get('/', response_model=schemas.Healthcheck)
+def Healthcheck():
     return {
         "up_since": start_stamp
     }

parser_api/router.py

@@ -10,24 +10,65 @@ from .database import SessionLocal, get_db
 router = APIRouter(prefix='/api')
 
 
-@router.get('/list', response_model=List[schemas.Record])
+@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])
+@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)
+@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)

parser_api/schemas.py

@@ -37,3 +37,6 @@ class RecordCreate(BaseRecord):
 class CheckResponse(BaseModel):
     is_outage: bool
     when_finish: Optional[datetime.datetime] = None
+
+class Healthcheck(BaseModel):
+    up_since: datetime.datetime

requirements.dev.txt

@@ -1,6 +1,10 @@
+aiohttp==3.8.5
+aiosignal==1.3.1
 annotated-types==0.5.0
 anyio==3.7.1
 asttokens==2.4.0
+async-timeout==4.0.3
+attrs==23.1.0
 autopep8==2.0.4
 backcall==0.2.0
 beautifulsoup4==4.12.2
@@ -14,6 +18,7 @@ decorator==5.1.1
 executing==1.2.0
 fastapi==0.103.1
 fonttools==4.42.1
+frozenlist==1.4.0
 greenlet==2.0.2
 idna==3.4
 ipykernel==6.25.2
@@ -25,6 +30,7 @@ kiwisolver==1.4.5
 lxml==4.9.3
 matplotlib==3.8.0
 matplotlib-inline==0.1.6
+multidict==6.0.4
 mypy==1.5.1
 mypy-extensions==1.0.0
 nest-asyncio==1.5.8
@@ -70,3 +76,4 @@ typing_extensions==4.8.0
 tzdata==2023.3
 urllib3==2.0.4
 wcwidth==0.2.6
+yarl==1.9.2

requirements.txt

@@ -1,15 +1,21 @@
+aiohttp==3.8.5
+aiosignal==1.3.1
 annotated-types==0.5.0
 anyio==3.7.1
+async-timeout==4.0.3
+attrs==23.1.0
 beautifulsoup4==4.12.2
 bs4==0.0.1
 certifi==2023.7.22
 charset-normalizer==3.2.0
 click==8.1.7
 fastapi==0.103.1
+frozenlist==1.4.0
 greenlet==2.0.2
 h11==0.14.0
 idna==3.4
 lxml==4.9.3
+multidict==6.0.4
 numpy==1.26.0
 pandas==2.1.1
 psycopg==3.1.10
@@ -30,3 +36,4 @@ typing_extensions==4.8.0
 tzdata==2023.3
 urllib3==2.0.5
 uvicorn==0.23.2
+yarl==1.9.2

rosseti_parser/README.md

@@ -29,10 +29,22 @@ class RossetiParser:
 ```python
 def split_addresses(df: pd.DataFrame) -> pd.DataFrame
 ```
+- `get_building_id`:
+```python
+def get_building_id(street: str) -> Tuple[Optional[int], Optional[float], Optional[float]]
+```
 - `fetch_builing_ids`:
 ```python
 def fetch_builing_ids(df: pd.DataFrame) -> pd.DataFrame
 ```
+- `async_fetch_building_ids`:
+```python
+async def async_fetch_building_ids(df: pd.DataFrame) -> pd.DataFrame
+```
+- `concurrent_fetch_builing_ids`:
+```python
+def concurrent_fetch_builing_ids(df: pd.Dataframe) -> pd.DataFrame
+```
 - `preprocess_df`:
 ```python
 def preprocess_df(df: pd.DataFrame) -> pd.DataFrame

rosseti_parser/__init__.py

@@ -1,5 +1,5 @@
 from .rosseti import RossetiParser
 from .address import split_addresses
-from .building_id import fetch_builing_ids
+from .building_id import get_building_id, fetch_builing_ids, async_fetch_building_ids, concurrent_fetch_builing_ids
 from .preprocess import preprocess_df, COL_NS, ICOL_NS, preprocess_read_df, group_by_index
 from .util import pipeline

rosseti_parser/building_id.py

@@ -4,6 +4,8 @@ from typing import Optional, Tuple, Any, List
 import requests
 import pandas as pd
 import numpy as np
+import asyncio
+import aiohttp
 
 GeoTupleType = Tuple[Optional[int], Optional[float], Optional[float]]
 
@@ -29,3 +31,42 @@ def fetch_builing_ids(df: pd.DataFrame) -> pd.DataFrame:
         lambda row: get_building_id(row['Улица']), axis=1, result_type='expand')
 
     return df
+
+
+async def async_fetch_building_id(session: aiohttp.ClientSession, street: str) -> GeoTupleType:
+    if pd.isnull(street):
+        return None, None, None
+
+    async with session.get('https://geocode.gate.petersburg.ru/parse/eas', params={
+        'street': street
+    }) as r:
+        res = await r.json()
+
+        if 'error' in res:
+            return None, None, None
+
+        return res['Building_ID'], res['Latitude'], res['Longitude']
+
+
+async def async_fetch_building_ids(df: pd.DataFrame) -> pd.DataFrame:
+    async with aiohttp.ClientSession() as session:
+        tasks = []
+
+        for _, row in df.iterrows():
+            tasks.append(
+                asyncio.ensure_future(
+                    async_fetch_building_id(session, row['Улица'])
+                )
+            )
+
+        res = await asyncio.gather(*tasks)
+
+        df[['ID здания', 'Широта', 'Долгота']] = res
+
+        return df
+
+
+def concurrent_fetch_builing_ids(df: pd.Dataframe) -> pd.DataFrame:
+    return asyncio.run(
+        async_fetch_building_ids(df)
+    )

rosseti_parser/util.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from . import RossetiParser, split_addresses, fetch_builing_ids, preprocess_df
+from . import RossetiParser, split_addresses, concurrent_fetch_builing_ids, preprocess_df
 
 
 def pipeline(parser: Optional[RossetiParser] = None) -> RossetiParser:
@@ -11,7 +11,7 @@ def pipeline(parser: Optional[RossetiParser] = None) -> RossetiParser:
 
     parser.df = split_addresses(parser.df)
 
-    parser.df = fetch_builing_ids(parser.df)
+    parser.df = concurrent_fetch_builing_ids(parser.df)
 
     parser.df = preprocess_df(parser.df)