gisaf-backend/src/gisaf/api/main.py

import logging
from datetime import timedelta
from typing import Annotated

from fastapi import Depends, APIRouter, HTTPException, status, responses
from sqlalchemy.orm import selectinload
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import select

from gisaf.models.authentication import (
  User, UserRead,
  Role, RoleRead,
)
from gisaf.models.category import Category, CategoryRead
from gisaf.models.to_migrate import DataProvider
from gisaf.models.survey import Equipment, SurveyMeta, Surveyor
from gisaf.config import Survey, conf
from gisaf.models.bootstrap import BootstrapData
from gisaf.models.store import Store, StoreNameOnly
from gisaf.models.project import Project
from gisaf.models.authentication import UserRoleLink #, ACL
from gisaf.database import pandas_query, fastapi_db_session as db_session
from gisaf.security import (
    Token,
    authenticate_user, get_current_user, create_access_token,
    )
from gisaf.registry import registry, NotInRegistry
from gisaf.custom_store_base import BaseStore
from gisaf.models.to_migrate import (
  FeatureInfo, InfoItem, Attachment, InfoCategory
)
from gisaf.live_utils import get_live_feature_info

logger = logging.getLogger(__name__)


api = APIRouter(
    tags=["api"],
    # dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)
#api.add_middleware(SessionMiddleware, secret_key=conf.crypto.secret)

@api.get('/bootstrap')
async def bootstrap(
    user: Annotated[UserRead, Depends(get_current_user)]) -> BootstrapData:
    return BootstrapData(user=user)


@api.post("/token")
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends()
    ) -> Token:
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(seconds=conf.crypto.expire))
    return Token(access_token=access_token, token_type='bearer')


@api.get("/users")
async def get_users(
    db_session: db_session,
    ) -> list[UserRead]:
    query = select(User).options(selectinload(User.roles)) # type: ignore[arg-type]
    data = await db_session.exec(query)
    return data.all() # type: ignore[return-value]

@api.get("/roles")
async def get_roles(
    db_session: db_session,
    ) -> list[RoleRead]:
    query = select(Role).options(selectinload(Role.users)) # type: ignore[arg-type]
    data = await db_session.exec(query)
    return data.all() # type: ignore[return-value]

@api.get('/acls')
async def get_acls(db_session: db_session,
    user: Annotated[User, Depends(get_current_user)]) -> list[UserRoleLink]:
    """New: ACLs returned as UserRoleLink"""
    if not user or not user.has_role('manager'):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    data = await db_session.exec(select(UserRoleLink))
    return data.all() # type: ignore[return-value]

@api.get("/categories")
async def get_categories(
    db_session: db_session,
    ) -> list[CategoryRead]:
    query = select(Category)
    data = await db_session.exec(query)
    return data.all() # type: ignore[return-value]

@api.get("/categories_pandas")
async def get_categories_p(
    db_session: db_session,
    ) -> list[CategoryRead]:
    query = select(Category)
    df = await db_session.run_sync(pandas_query, query)
    return df.to_dict(orient="records") # type: ignore[return-value]

# @api.get("/list")
@api.get("/data-providers")
async def list_data_providers() -> list[DataProvider]:
    """
    Return a list of data providers, for use with the api (graphs, etc)
    :return:
    """
    return [
        DataProvider(
            name=model.get_store_name(),
            values=[value.get_store_name() for value in values]
        ) for model, values in registry.values_for_model.items()]

@api.get("/stores")
async def get_stores() -> list[Store]:
    df = registry.stores.reset_index().\
        drop(columns=['model', 'raw_model', 'base_gis_type'])
    return df.to_dict(orient="records") # type: ignore[return-value]

@api.get("/projects")
async def get_projects(
    db_session: db_session,
    ) -> list[Project]:
    query = select(Project)
    data = await db_session.exec(query)
    return data.all() # type: ignore[return-value]

@api.get("/survey_meta")
async def get_survey_meta(
    db_session: db_session,
    ) -> SurveyMeta:
    return SurveyMeta(
        projects=(await db_session.exec(select(Project))).all(), # type: ignore[arg-type]
        surveyors=(await db_session.exec(select(Surveyor))).all(), # type: ignore[arg-type]
        equipments=(await db_session.exec(select(Equipment))).all(), # type: ignore[arg-type]
        statuses=conf.map.status,
        stores_misc=[StoreNameOnly(name=name)
                     for name, model in registry.geom_custom.items()],
        stores_line_work=[StoreNameOnly(name=name)
                          for name in registry.stores[registry.stores.is_line_work].index],
        default=conf.admin.basket.default
    )

@api.get("/feature-info/{store}/{id}")
async def get_feature_info(
    store: str, id: str,
    ) -> FeatureInfo:
    if store not in registry.stores.index:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    store_record = registry.stores.loc[store]
    model = store_record.model
    if store_record.is_live:
        feature_info = await get_live_feature_info(store, id)
    elif issubclass(model, BaseStore):
        feature_info = await model.get_item_params(id)
    else:
        ## A layer in the database
        try:
            feature_info = await registry.get_model_id_params(model, int(id))
        except NotInRegistry:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return feature_info

# @api.get("/user-role")
# async def get_user_role_relation(
#     *, db_session: AsyncSession = Depends(get_db_session)
#     ) -> list[UserRoleLink]:
#     roles = await db_session.exec(select(UserRoleLink))
#     return roles.all()