gisaf-backend/src/gisaf/api.py

113 lines
3.3 KiB
Python
Raw Normal View History

import logging
2023-11-06 17:04:17 +05:30
from datetime import timedelta
from typing import Annotated
2023-11-06 17:04:17 +05:30
from fastapi import Depends, FastAPI, HTTPException, status, responses
from sqlalchemy.orm import selectinload
2023-11-06 17:04:17 +05:30
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from .models.authentication import (
User, UserRead,
Role, RoleRead,
)
from .models.category import Category, CategoryRead
from .config import conf
from .models.bootstrap import BootstrapData
from .models.store import Store
2023-11-06 17:04:17 +05:30
from .database import get_db_session, pandas_query
from .security import (
Token,
authenticate_user, get_current_user, create_access_token,
2023-11-06 17:04:17 +05:30
)
from .registry import registry
2023-11-06 17:04:17 +05:30
logger = logging.getLogger(__name__)
2023-11-06 17:04:17 +05:30
api = FastAPI(
default_response_class=responses.ORJSONResponse,
)
#api.add_middleware(SessionMiddleware, secret_key=conf.crypto.secret)
2023-11-06 17:04:17 +05:30
db_session = Annotated[AsyncSession, Depends(get_db_session)]
2023-11-06 17:04:17 +05:30
@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:
2023-11-06 17:04:17 +05:30
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))
2023-11-06 17:04:17 +05:30
return {"access_token": access_token, "token_type": "bearer"}
@api.get("/list")
async def list_data_providers():
"""
Return a list of data providers, for use with the api (graphs, etc)
:return:
"""
return [{'name': m.__name__, 'store': m.get_store_name()}
for m in registry.values_for_model]
2023-11-06 17:04:17 +05:30
@api.get("/users")
async def get_users(
db_session: db_session,
2023-11-06 17:04:17 +05:30
) -> 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,
2023-11-06 17:04:17 +05:30
) -> 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,
2023-11-06 17:04:17 +05:30
) -> list[CategoryRead]:
query = select(Category)
data = await db_session.exec(query)
return data.all()
@api.get("/categories_pandas")
2023-11-06 17:04:17 +05:30
async def get_categories_p(
db_session: db_session,
2023-11-06 17:04:17 +05:30
) -> list[CategoryRead]:
query = select(Category)
df = await db_session.run_sync(pandas_query, query)
return df.to_dict(orient="records")
@api.get("/stores")
async def get_stores() -> list[Store]:
df = registry.stores.reset_index().drop(columns=['model', 'raw_model'])
return df.to_dict(orient="records")
2023-11-06 17:04:17 +05:30
# @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()