[How to use] I'm not able to use ST_* functions during insertion about geoalchemy2

[adrien-berchet]

Hi @rmmariano
I just took a quick look but as far I can see you mixed core and ORM queries. If you want to use core queries, you can look at https://geoalchemy-2.readthedocs.io/en/latest/gallery/test_length_at_insert.html#sphx-glr-gallery-test-length-at-insert-py, while if you want to use ORM queries you can look at https://geoalchemy-2.readthedocs.io/en/latest/gallery/test_type_decorator.html#sphx-glr-gallery-test-type-decorator-py.
Basically, with core queries you should use func.ST_* and bindparam in the insert queries while with ORM queries you should create a specific type that will insert the proper func.ST_* automatically at insert.

[rmmariano]

Hi @adrien-berchet
Thank you for you awnswer.
I've read the example, but it's not clear to me how I could update my example to use ORM.
I'm newbie on geoalchemy.
Could you please send my an example based on the one that I made? Using ORM, when you have free time.
Thanks a lot.

[adrien-berchet]

Hi @rmmariano
Here is an example of what I spoke about. As I said, you have to create a specific type for your bbox column that automatically use ST_Envelope at insert. I also added an example with core inserts in the end. I hope it will solve your issue.

from geoalchemy2 import Geometry, WKBElement, WKTElement
from sqlalchemy import Integer, create_engine, bindparam, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from sqlalchemy.sql import func
from sqlalchemy.types import TypeDecorator


db_connection_url = "postgresql://gis:gis@localhost/gis"
engine = create_engine(url=db_connection_url, echo=True)


class BBoxGeometry(TypeDecorator):
    """This class is used to insert a ST_Force3D() in each insert."""

    impl = Geometry

    cache_ok = True

    def bind_expression(self, bindvalue):
        return func.ST_Envelope(
            self.impl.bind_expression(bindvalue),
            type=self,
        )

class Base(DeclarativeBase):
    ...

class Geofence(Base):
    __tablename__ = "geofence"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    geom: Mapped[WKBElement] = mapped_column(
        Geometry(geometry_type="POLYGON", srid=4326), nullable=False
    )
    bbox: Mapped[WKBElement] = mapped_column(
        BBoxGeometry(geometry_type="POLYGON", srid=4326), nullable=False
    )

Base.metadata.drop_all(engine, checkfirst=True)
Base.metadata.create_all(engine)

# 

raw_geom = 'POLYGON ((0 0, 1 1, 2 0, 1 -1, 0 0))'
geom = WKTElement(raw_geom)

geofences = [
    Geofence(
        id=1,
        geom=geom,
        bbox=geom,
    ),
    Geofence(
        id=2,
        geom=geom,
        bbox=geom,
    )
]

Session = sessionmaker(bind=engine, expire_on_commit=False)

with Session() as session:
    # With ORM
    try:
        session.bulk_save_objects(geofences)
    except Exception:
        session.rollback()
        raise
    else:
        session.commit()

    # With core
    conn = session.connection()
    i = Geofence.__table__.insert()
    i = i.values(id=bindparam("id"), geom=bindparam("geom"), bbox=func.ST_Envelope(func.ST_GeomFromEWKT(bindparam("geom"))))
    conn.execute(i, [{"id": j.id + 10, "geom": j.geom, "bbpx": j.bbox} for j in geofences])

    # Check the inserted results
    res = conn.execute(select(Geofence.__table__.c.id, Geofence.__table__.c.geom.ST_AsText(), Geofence.__table__.c.bbox.ST_AsText())).all()
    for j in res:
        print(j)