from typing import Any, ClassVar from pydantic import computed_field, ConfigDict from sqlmodel import Field, Relationship, SQLModel, JSON, TEXT, Column, select from .metadata import gisaf_survey from ..database import db_session, pandas_query mapbox_type_mapping = { 'Point': 'symbol', 'Line': 'line', 'Polygon': 'fill', } class BaseModel(SQLModel): @classmethod async def get_df(cls): async with db_session() as session: query = select(cls) return await session.run_sync(pandas_query, query) class CategoryGroup(BaseModel, table=True): metadata = gisaf_survey __tablename__ = 'category_group' name: str | None = Field(min_length=4, max_length=4, default=None, primary_key=True) major: str long_name: str categories: list['Category'] = Relationship(back_populates='category_group') class Admin: menu = 'Other' flask_admin_model_view = 'CategoryGroupModelView' class CategoryModelType(BaseModel, table=True): metadata = gisaf_survey __tablename__ = 'category_model_type' name: str = Field(default=None, primary_key=True) class Admin: menu = 'Other' flask_admin_model_view = 'MyModelViewWithPrimaryKey' class CategoryBase(BaseModel): model_config = ConfigDict(protected_namespaces=()) class Admin: menu = 'Other' flask_admin_model_view = 'CategoryModelView' name: str | None = Field(default=None, primary_key=True) domain: ClassVar[str] = 'V' description: str | None group: str = Field(min_length=4, max_length=4, foreign_key="category_group.name", index=True) minor_group_1: str = Field(min_length=4, max_length=4, default='----') minor_group_2: str = Field(min_length=4, max_length=4, default='----') status: str = Field(min_length=1, max_length=1) custom: bool | None auto_import: bool = True model_type: str = Field(max_length=50, foreign_key='category_model_type.name', default='Point') long_name: str | None = Field(max_length=50) style: str | None = Field(sa_type=TEXT) symbol: str | None = Field(max_length=1) mapbox_type_custom: str | None = Field(max_length=32) mapbox_paint: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True)) mapbox_layout: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True)) viewable_role: str | None extra: dict[str, Any] | None = Field(sa_type=JSON(none_as_null=True)) @computed_field @property def layer_name(self) -> str: """ ISO compliant layer name (see ISO 13567) :return: str """ return '{self.domain}-{self.group:4s}-{self.minor_group_1:4s}-{self.minor_group_2:4s}-{self.status:1s}'.format(self=self) @computed_field @property def table_name(self) -> str: """ Table name :return: """ if self.minor_group_2 == '----': return '{self.domain}_{self.group:4s}_{self.minor_group_1:4s}'.format(self=self) else: return '{self.domain}_{self.group:4s}_{self.minor_group_1:4s}_{self.minor_group_2:4s}'.format(self=self) @computed_field @property def raw_survey_table_name(self) -> str: """ Table name :return: """ if self.minor_group_2 == '----': return 'RAW_{self.domain}_{self.group:4s}_{self.minor_group_1:4s}'.format(self=self) else: return 'RAW_{self.domain}_{self.group:4s}_{self.minor_group_1:4s}_{self.minor_group_2:4s}'.format(self=self) @computed_field @property def mapbox_type(self) -> str: return self.mapbox_type_custom or mapbox_type_mapping[self.model_type] class Category(CategoryBase, table=True): metadata = gisaf_survey name: str = Field(default=None, primary_key=True) category_group: CategoryGroup = Relationship(back_populates="categories") class CategoryRead(CategoryBase): name: str