sqlalchemy hstore test

gistfile1.txt

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext import mutable as sa_mutable

import sqlalchemy.dialects.postgresql

import string

import pdb

Base = declarative_base()
engine = create_engine('postgresql://sa_test:sa_test@localhost/sqlalchemy_test', echo=True)



class User(Base): 
    __tablename__     = 'user_data' 
    user_id = Column('record_id', Integer, primary_key=True) 
    login = Column('login', String(12), nullable=False) 
    prefs =  Column('prefs', sa_mutable.MutableDict.as_mutable(sqlalchemy.dialects.postgresql.HSTORE) , nullable=True )


Base.metadata.create_all(engine) 
dbSessionmaker = sessionmaker(bind=engine)
dbSession = dbSessionmaker()



if False :
    ## only need to run this once...

    ## remember to first run in this db as pg superuser -- 
    sql_prefix = """CREATE EXTENSION HSTORE;"""
    sql_create = """CREATE TABLE user_data ( 
        record_id SERIAL PRIMARY KEY , 
        login VARCHAR(12), 
        prefs HSTORE 
    );"""
    


    people = [ 'jonathan', 'michael', 'adam' ]
    for person in people :
        user = User()
        user.login = person
        user.prefs = {'inboundBusStop':None}
        dbSession.add(user)
        dbSession.flush()
    dbSession.commit()
    



if True:
    results_a = dbSession.query( User )\
                .filter(\
                    User.prefs.has_key('inboundBusStop')
                )\
                .all()
    print "results_a ->"
    for i in results_a:
        print "\t%s | %s | %s" % ( i.user_id , i.login , i.prefs )


    try:
        print "results_b ->"
        
        ## WRITE YOUR ROUTINES HERE
        
    except: 
        print "\t b = bad"
        
        
    print "results_c ->"
    results_c = dbSession.execute(
        User.__table__\
        .update()\
        .values( 
            prefs = User.__table__.c.prefs + sqlalchemy.dialects.postgresql.hstore(sqlalchemy.dialects.postgresql.array(['transitAvaiable', 'true'])) 
        )\
        .where(\
            User.prefs.has_key('inboundBusStop')
        )
    )    
    print "works!"