pg_custom_types.py

# How to use PG Point type in SQAlchemy
#
# Most of the code comes from here:
# http://initd.org/psycopg/docs/advanced.html#adapting-new-types

import re

import sqlalchemy
import psycopg2
from psycopg2.extensions import adapt, register_adapter, AsIs


class Point():
    def __init__(self, x, y):
        self.x = x
        self.y = y


def adapt_point(point):
    return AsIs("'(%s, %s)'" % (adapt(point.x), adapt(point.y)))


def cast_point(value, cur):
    if value is None:
        return None

    # Convert from (f1, f2) syntax using a regular expression.
    m = re.match(r"\(([^)]+),([^)]+)\)", value)
    if m:
        return Point(float(m.group(1)), float(m.group(2)))
    else:
        raise psycopg2.InterfaceError("bad point representation: %r" % value)


def register_point_type(engine):
    register_adapter(Point, adapt_point)
    rs = engine.execute("SELECT NULL::point")
    point_oid = rs.cursor.description[0][1]
    POINT = psycopg2.extensions.new_type((point_oid,), "POINT", cast_point)
    psycopg2.extensions.register_type(POINT)


engine = sqlalchemy.create_engine(
    'postgresql+psycopg2://lbolla@localhost/mytest')

register_point_type(engine)

if __name__ == '__main__':
    engine.execute('''
        create table t (
            p point
        )
    ''')

    engine.execute("insert into t values (%s)", Point(1, 2))
    print(dict(engine.execute("select * from t").fetchone()))

    engine.execute('drop table t')