sqlalchemy_geom.py

# -*- coding: utf-8 -*-
import json

import sqlalchemy as sa

from sqlalchemy.types import UserDefinedType

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Geometry(UserDefinedType):
    def get_col_spec(self):
        return "GEOMETRY"

    def bind_expression(self, bindvalue):
        return sa.func.ST_GeomFromGeoJSON(bindvalue, type_=self)

    def column_expression(self, col):
        return sa.func.ST_AsGeoJSON(col, type_=self)

    def result_processor(self, dialect, coltype):
        def process(value):
            if value is not None:
                return json.loads(value)
            return None
        return process

    def bind_processor(self, dialect):
        def process(value):
            if value is not None:
                return json.dumps(value)
            return None
        return process



class TableWithGeom(Base):
    __tablename__ = 'with_geom'

    id = sa.Column(sa.Integer, primary_key=True)
    geom = sa.Column('the_geom', Geometry)