-
Notifications
You must be signed in to change notification settings - Fork 112
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Docs: Add new mapping style example in the gallery (#473)
- Loading branch information
1 parent
0d81f70
commit 829567b
Showing
1 changed file
with
61 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,61 @@ | ||
""" | ||
New ORM Declarative Mapping Style | ||
================================= | ||
``SQLAlchemy>=2`` introduced a new way to construct mappings using the | ||
``sqlalchemy.orm.DeclarativeBase`` base class. | ||
This example shows how to use GeoAlchemy2 types in this context. | ||
""" | ||
import pytest | ||
from pkg_resources import parse_version | ||
from sqlalchemy import __version__ as SA_VERSION | ||
|
||
try: | ||
from sqlalchemy.orm import DeclarativeBase | ||
from sqlalchemy.orm import Mapped | ||
from sqlalchemy.orm import mapped_column | ||
except ImportError: | ||
pass | ||
|
||
from geoalchemy2 import Geometry | ||
from geoalchemy2 import WKBElement | ||
from geoalchemy2 import shape | ||
|
||
|
||
def check_wkb(wkb, x, y): | ||
pt = shape.to_shape(wkb) | ||
assert round(pt.x, 5) == x | ||
assert round(pt.y, 5) == y | ||
|
||
|
||
@pytest.mark.skipif( | ||
parse_version(SA_VERSION) < parse_version("2"), | ||
reason="New ORM mapping is only available for sqlalchemy>=2", | ||
) | ||
def test_ORM_mapping(session, conn, schema): | ||
class Base(DeclarativeBase): | ||
pass | ||
|
||
class Lake(Base): | ||
__tablename__ = "lake" | ||
__table_args__ = {"schema": schema} | ||
id: Mapped[int] = mapped_column(primary_key=True) | ||
mapped_geom: Mapped[WKBElement] = mapped_column(Geometry(geometry_type="POINT", srid=4326)) | ||
|
||
Lake.__table__.drop(conn, checkfirst=True) | ||
Lake.__table__.create(bind=conn) | ||
|
||
# Create new point instance | ||
p = Lake() | ||
p.mapped_geom = "SRID=4326;POINT(5 45)" | ||
|
||
# Insert point | ||
session.add(p) | ||
session.flush() | ||
session.expire(p) | ||
|
||
# Query the point and check the result | ||
pt = session.query(Lake).one() | ||
assert pt.id == 1 | ||
assert pt.mapped_geom.srid == 4326 | ||
check_wkb(pt.mapped_geom, 5, 45) |