Nick Doiron / Oct 14 2019
Remix of Python by Nextjournal

PyCon India - GeoAlchemy2

import sys; sys.version.split()[0]
  • Let's install SQLAlchemy and GeoAlchemy2 - note that '2' is part of the name, and not a version number:
pip install sqlalchemy geoalchemy2 psycopg2-binary
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from geoalchemy2 import Geography

Let's connect to this PostgreSQL database, where I have already installed PostGIS, and created a read-only user py:zim.

from sqlalchemy import create_engine
engine = create_engine('postgresql://', echo=True)
  • With SQLAlchemy, we can use a class to represent each column of the table.

GeoAlchemy2 gives us the Geography type and additional functions.

Base = declarative_base()
class State(Base):
  __tablename__ = 'states'
  ogc_fid = Column(Integer, primary_key=True)
  st_nm = Column(String)
  wkb_geometry = Column(Geography)
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

SELECT-ing the geo column

Here we call State.wkb_geometry, and it returns a spatial object

mystate = session.query(State.wkb_geometry).first()
  • It's better to use a new function such as ST_AsText() or ST_AsGeoJSON() to return a usable object:
mystate = session.query(State.wkb_geometry.ST_AsText()).first()

Spatial + COUNT

class Tower(Base):
  __tablename__ = 'cell_towers'
  ID = Column(Integer, primary_key=True)
  point = Column(Geography)

class Bengaluru(Base):
  __tablename__ = 'bengaluru'
  ogc_fid = Column(Integer, primary_key=True)
  wkb_geometry = Column(Geography)

Spatial + JOIN

class Ward(Base):
  __tablename__ = 'chennai_wards'
  ogc_fid = Column(Integer, primary_key=True)
  wkb_geometry = Column(Geography)
  ward_no = Column(Integer)
  zone_no = Column(Integer)
class Parliament(Base):
  __tablename__ = 'parliament'
  ogc_fid = Column(Integer, primary_key=True)
  wkb_geometry = Column(Geography)
  pc_name = Column(String)
  pc_code = Column(Integer)

Here we do not do a JOIN outright, but SELECT from both tables and find all matches:

combined = session.query(Ward, Parliament).filter(Ward.wkb_geometry.ST_Intersects(Parliament.wkb_geometry)).order_by(Ward.ward_no).all()
for w, p in combined:
  print ("Ward: {} in Parliament: {}".format(w.ward_no, p.pc_name))