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

[arashmad]

Here is my workaround:

import uuid
from datetime import datetime, timezone

from geoalchemy2 import Geometry
from geoalchemy2.shape import from_shape

from sqlmodel import SQLModel, Field, create_engine, Column, Session


def generate_id(unique_string: str) -> str:
 """Generate unique id."""
 return str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{unique_string}"))

class MyTable(SQLModel, table=True):
 """Model to store spatial data."""
 record_id: str = Field(
 title="Record ID",
 description="Unique identifier for the record",
 primary_key=True
 )
 footprint: bytes | None = Field(
 title="Footprint",
 description="Footprint of the data",
 sa_column=Column(Geometry("POLYGON", srid=4326))
 )
 acquisition_date: datetime = Field(
 title="Record Date",
 description="Timestamp in UTC when data was acquired",
 )

 def __init__(self, **kwargs):
 super().__init__(**kwargs)
 if not self.record_id and self.acquisition_date:
 self.record_id = \
 generate_id(str(self.acquisition_date))

# Database connection and initiation
DB_CONN_URL = f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
engine = create_engine(DB_CONN_URL, echo=True)
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)

# Create geometry object
footprint= box(long_min, lat_min, long_max, lat_max)
footprint_wkb_geom = from_shape(footprint, srid=4326)

# Create a data entry
data_entry = MyTable(
	footprint=ga.functions.ST_GeomFromEWKB(bounding_box_wkb_geom),
	acquisition_date=acquisition_date)

# Insert into table
with Session(engine) as session:
	session.add(data_entry)
	session.commit()
	session.refresh(data_entry)

[lionpeloux]

Hey @arashmad , could you provide a full MWE ? I would really appreciate.

I could not reproduce your solution.
box and from_shape are not defined ...
Are you able to read AND write to the DB ?

Many thanks !

[lionpeloux]

Ok, I came up with a mode complete MWE, I think it could be usefull for others :

import uuid
from typing import Any

import geoalchemy2 as ga
from geoalchemy2 import Geometry
from geoalchemy2.shape import from_shape, to_shape
from pydantic import computed_field, model_validator
from shapely.geometry import Point as ShapelyPoint
from sqlalchemy import Column
from sqlmodel import Field, Session, SQLModel, create_engine, select

# https://github.com/fastapi/sqlmodel/issues/528#issuecomment-2626806093
# https://github.com/fastapi/sqlmodel/issues/453
# https://github.com/fastapi/sqlmodel/pull/1041


def get_db_url(login: str = "postgres", password: str = "postgres"):
 import urllib.parse

 return f"postgresql://{urllib.parse.quote(login)}:{urllib.parse.quote(password)}@127.0.0.1:54322/postgres"


class Postgis(SQLModel, table=True, table_name="postgis"):
 """Model to store spatial data."""

 id: str = Field(
 default_factory=lambda: str(uuid.uuid4()),
 primary_key=True,
 )

 point: Any | None = Field(
 sa_column=Column(Geometry("POINT", srid=4326)), exclude=True
 )

 @computed_field # type: ignore[prop-decorator]
 @property
 def latitude(self) -> float | None:
 pt = to_shape(self.point) # type: ignore
 return float(pt.y)

 @computed_field # type: ignore[prop-decorator]
 @property
 def longitude(self) -> float | None:
 pt = to_shape(self.point) # type: ignore
 return float(pt.x)

 @model_validator(mode="before")
 @classmethod
 def coerce_point(cls, data: Any) -> Any:
 # propose various ways to provide the point data
 # could be improved + better error handling
 lat = data.get("latitude")
 lon = data.get("longitude")
 if not lat or not lon:
 raise ValueError(
 "latitude and longitude must be provided as arguments to model_validate"
 )
 data["point"] = ga.functions.ST_GeomFromEWKB(
 from_shape(ShapelyPoint(lon, lat), srid=4326)
 )
 return data


engine = create_engine(get_db_url(), echo=True)

# create test table
metadata = SQLModel.metadata
postgis_table = metadata.tables["postgis"]
metadata.drop_all(engine, tables=[postgis_table])
metadata.create_all(engine, tables=[postgis_table])

# create test data
# you must use model_validate, not Postgis constructor as Pydantic validation is 
# disabled for SQLModel with table=true
# see : https://github.com/fastapi/sqlmodel/issues/453 
latitude = 48.8156
longitude = 2.2241
data_entry = Postgis.model_validate(
 {
 "latitude": latitude,
 "longitude": longitude,
 }
)

# Insert into table
with Session(engine) as session:
 session.add(data_entry)
 session.commit()
 session.refresh(data_entry)
 inserted_row = data_entry.model_dump_json(indent=2)
 print(inserted_row)
 
# Select from table
with Session(engine) as session:
 statement = select(Postgis)
 result = session.exec(statement).one()
 selected_row = result.model_dump_json(indent=2)
 print(selected_row)


# Compare the two rows

assert inserted_row == selected_row, "data_entry and result should be the same"

[lionpeloux]

An other approach is to have X, Y , Z columns with an automatic POINT column.
In SQLModel you would only provide a "partial" table (omitting to deal with point).
This is not working if you rely on SQLModel for db creation.

CREATE TABLE addresses (
 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
 street_name TEXT NOT NULL,
 street_number TEXT,
 city TEXT NOT NULL,
 postal_code TEXT NOT NULL,
 longitude DOUBLE PRECISION,
 latitude DOUBLE PRECISION,
 elevation DOUBLE PRECISION,
 point EXTENSIONS.GEOMETRY(POINT, 4326) GENERATED ALWAYS AS (
 CASE 
 WHEN longitude IS NOT NULL AND latitude IS NOT NULL 
 THEN ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)
 ELSE NULL
 END
 ) STORED,
 created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
 updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

[arashmad]

@lionpeloux Thanks for sharing it.
I saw you already found your own solution.
When I have time, I will try it out to see your fix.