ARRAY of POINT from PostgreSQL in SQLModel

[ddzmanashvili]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from sqlmodel import Field, Session, SQLModel, create_engine, select, ARRAY, Integer, String, FLOAT
from geoalchemy2 import Geometry
from typing import Optional
from datetime import datetime
from urllib.parse import quote_plus
from typing import List, Optional, Set
from sqlalchemy.sql.schema import Column



class Site_Metrics(SQLModel, table=True):
    site_metric_id: Optional[int] = Field(default=None, primary_key=True)
    site_id: int
    metric_id: int
    created_at: datetime
    updated_at: datetime
    n_value: float
    a_value: Optional[List] = Field(default_factory=list, sa_column=Column(ARRAY(Geometry('POINT'))))
    deleted_at: datetime

Description

I have point[] type in SQLModel but by far without luck. I tried many things but this variation is the closest I have got. Is there any other way to do this? Thanks

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.8

Python Version

3.8.0

Additional Context

No response

[saschahofmann]

Any success with this? I also tried to find whether pydantic has some capabilities around geospatial data. The closest I have found is https://pypi.org/project/geojson-pydantic/

[saschahofmann]

I managed to parse the DB data to a json response by defining this

from pydantic_geojson import FeatureModel
from geoalchemy2 import WKBElement
from geoalchemy2.shape import to_shape
from shapely import to_geojson
import json

class WKBToGeoJSON(FeatureModel):
    @classmethod
    def validate(cls, value):
        if not isinstance(value, WKBElement):
            raise TypeError("value must be a WKBElement object")
        shape = to_shape(value)
        if shape.type != "Point":
            raise TypeError("value must be a Point")
        return json.loads(to_geojson(shape))

And then defining my model using this Field

class Asset(BaseModel, table=True):
    name: Optional[str] = Field(default=None)
    description: Optional[str] = Field(default=None)
    location: WKBToGeoJSON = Field(
        default=None, sa_column=Column(Geometry("POINT", srid=4326))
    )

I havent tried it yet but I assume that the other way (geojson to sqlalchemy will fail).

@tiangolo is there some documentation on how to define the two way transformations for more exotic types?

[John-P]

I would also love to know how to handle geoJSON input/output for Geometry in a Postgres/Spatialite database backend. It appears to work just fine for WKT but it would be great to be able to validate/dump geoJSON. Two killer features for me are supporting geometry and JSON (e.g. for a JSONB column).

[John-P]

I'm not sure if this is the correct way to go about it, but I ended up doing this to test reading and writing geometry (using Postgres) as GeoJSON, and it appears to work well. If you are not using Postgres, you could replace the JSONB with regular SQLalchemy JSON.

If anyone knows of a better way to do this, please let me know.

import uuid
from typing import Literal, Optional

from fastapi import FastAPI, HTTPException

import geoalchemy2 as ga
from sqlmodel import Field, Session, SQLModel, create_engine, select
from sqlalchemy.dialects.postgresql import JSONB


engine = create_engine("postgresql://postgres:password@postgres/mydb")
app = FastAPI()


class GeoJSONGeometry(SQLModel):
    """A GeoJSON geometry fragment."""

    type: Literal["Point", "LineString", "Polygon"]
    coordinates: tuple[float, float] | list[tuple[float, float]] | list[list[tuple[float, float]]]


class Feature(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(
        default_factory=uuid.uuid4,
        primary_key=True,
        index=True,
        nullable=False,
    )
    geometry: bytes = Field(
        sa_type=ga.Geometry,
        nullable=False,
    )
    properties: Optional[dict] = Field(
        default=None,
        sa_type=JSONB,
        nullable=True,
    )


class FeatureCreate(SQLModel):
    geometry: GeoJSONGeometry = Field(
        default=None,
        sa_type=JSONB,
        nullable=False,
    )
    properties: Optional[dict]


class FeatureRead(FeatureCreate):
    id: uuid.UUID
    geometry: GeoJSONGeometry
    properties: Optional[dict]


@app.post("/features/", response_model=FeatureRead)
def create_feature(feature: FeatureCreate):
    with Session(engine) as session:
        db_feature = Feature(
            geometry=ga.functions.ST_SetSRID(
                ga.functions.ST_GeomFromGeoJSON(feature.geometry.model_dump_json()),
                0,  # PostGIS >3.0 ST_GeomFromGeoJSON sets the srid to 4326 by default
            ),
            properties=feature.properties,
        )
        session.add(db_feature)
        session.commit()
        session.refresh(db_feature)
        geojson = session.exec(
            select(ga.functions.ST_AsGeoJSON(db_feature.geometry))
        ).one()
    response = Feature(
        id=db_feature.id,
        geometry=GeoJSONGeometry.model_validate_json(geojson),
        properties=feature.properties,
    )
    return response


@app.get("/features/{feature_id}", response_model=FeatureRead)
def read_feature(feature_id: uuid.UUID):
    with Session(engine) as session:
        db_feature = session.get(Feature, feature_id)
        if db_feature is None:
            raise HTTPException(status_code=404, detail="Feature not found")
        geojson = session.exec(
            select(ga.functions.ST_AsGeoJSON(db_feature.geometry))
        ).one()
    response = FeatureRead(
        id=db_feature.id,
        geometry=GeoJSONGeometry.model_validate_json(geojson),
        properties=db_feature.properties,
    )
    return response