gisaf-backend/src/api.py

import logging
from datetime import timedelta
from time import time
from uuid import uuid1
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status, Request
from fastapi.security import OAuth2PasswordRequestForm
from starlette.middleware.sessions import SessionMiddleware

from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from sqlalchemy.orm import selectinload, joinedload

from .models.authentication import (
  User, UserRead,
  Role, RoleRead,
  UserRoleLink
)
from .models.category import (
  CategoryGroup, CategoryModelType,
  Category, CategoryRead
)
from .models.bootstrap import BootstrapData
from .database import get_db_session, pandas_query
from .security import (
    User as UserAuth,
    Token,
    authenticate_user, get_current_user, create_access_token,
    )
from .config import conf

logger = logging.getLogger(__name__)

api = FastAPI()
api.add_middleware(SessionMiddleware, secret_key=conf.crypto.secret)

db_session = Annotated[AsyncSession, Depends(get_db_session)]

@api.get("/nothing")
async def get_nothing() -> str:
    return ''


@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(
    db_session: db_session,
    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 {"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))
    data = await db_session.exec(query)
    return data.all()

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

@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()


@api.get("/categories_p")
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")

# @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()