philorg/gisaf-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages 6 Wiki Activity Actions

Cosmetic: mostly typings

Browse source
This commit is contained in:
phil 2024-02-13 12:47:07 +05:30
parent df5f67b79d
commit c1f229f805
13 changed files with 120 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

50
src/gisaf/api/v2.py
View file

@ -12,11 +12,14 @@ from gisaf.models.authentication import (
Role, RoleRead,
)
from gisaf.models.category import Category, CategoryRead
from gisaf.models.geo_models_base import LineWorkSurveyModel
from gisaf.models.to_migrate import DataProvider
from gisaf.config import conf
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
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,
@ -69,17 +72,26 @@ async def login_for_access_token(
async def get_users(
db_session: db_session,
) -> list[UserRead]:
query = select(User).options(selectinload(User.roles))
query = select(User).options(selectinload(User.roles)) # type: ignore[arg-type]
data = await db_session.exec(query)
return data.all()
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))
query = select(Role).options(selectinload(Role.users)) # type: ignore[arg-type]
data = await db_session.exec(query)
return data.all()
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(
@ -87,7 +99,7 @@ async def get_categories(
) -> list[CategoryRead]:
query = select(Category)
data = await db_session.exec(query)
return data.all()
return data.all() # type: ignore[return-value]
@api.get("/categories_pandas")
async def get_categories_p(
@ -95,7 +107,7 @@ async def get_categories_p(
) -> list[CategoryRead]:
query = select(Category)
df = await db_session.run_sync(pandas_query, query)
return df.to_dict(orient="records")
return df.to_dict(orient="records") # type: ignore[return-value]
# @api.get("/list")
@api.get("/data-providers")
@ -114,15 +126,31 @@ async def list_data_providers() -> list[DataProvider]:
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")
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)
df = await db_session.run_sync(pandas_query, query)
return df.to_dict(orient="records")
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(