sqlalchemy_demo.py

import logging
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy.orm import relation, mapper, sessionmaker
from sqlalchemy.sql import *
from sqlalchemy.ext.associationproxy import association_proxy

log = logging.getLogger('nm_test')
logging.basicConfig(level=logging.DEBUG,
        format='%(asctime)s,%(msecs)03d %(levelname)s [%(filename)s.%(funcName)s @ %(lineno)d.%(thread)d] %(message)s')

engine = create_engine('sqlite:///:memory:', echo = False, echo_pool = False)
meta = MetaData()
meta.bind = engine

developer_table = Table('developer', meta,
    Column('id', Integer, primary_key=True, autoincrement = False),
    Column('name', String),
)

project_table = Table('project', meta,
    Column('id', Integer, primary_key=True, autoincrement = True),
    Column('name', String)
)

developer_project_table = Table('developer_project', meta,
    Column('developer_id', Integer, ForeignKey('developer.id'), primary_key = True),
    Column('project_id', Integer, ForeignKey('project.id'), primary_key = True),
    Column('status', String)
)

class Developer(object):
    projects = association_proxy('developerProjects', 'projects')
    def __str__(self):
        return 'Developer id:%i, name:%s' % (self.id, self.name)

class Project(object):
    developers = association_proxy('developerProjects', 'developers')
    def __str__(self):
        return 'Project id:%i, name:%s' % (self.id, self.name)

class DeveloperProject(object):
    def __str__(self):
        return 'DeveloperProject developer:%s, project:%s, status:%s' % (self.developer_id, self.project_id, self.status)

mapper(Developer, developer_table, properties = {
    'developerProjects':relation(DeveloperProject, backref = "developers")
})

mapper(Project, project_table, properties = {
    'developerProjects':relation(DeveloperProject, backref = "projects")
})

mapper(DeveloperProject, developer_project_table)

meta.create_all(engine)
conn = engine.connect()

conn.execute(project_table.insert(),[
    {'name':'stackoverflow'},
    {'name':'superuser'},
])
conn.execute(developer_table.insert(),[
    {'name':'John'},
    {'name': 'TerryJ'},
    {'name': 'TerryG'},
    {'name': 'Eric'},
    {'name': 'Graham'},
])
conn.execute(developer_project_table.insert(),[
    {'developer_id':1, 'project_id':1, 'status':'active'},
    {'developer_id':2, 'project_id':2, 'status':'inactive'},
    {'developer_id':3, 'project_id':2, 'status':'active'},
    {'developer_id':4, 'project_id':1, 'status':'active'},
    {'developer_id':4, 'project_id':2, 'status':'active'},
    {'developer_id':5, 'project_id':1, 'status':'active'},
    {'developer_id':5, 'project_id':2, 'status':'inactive'},
])

Session = sessionmaker(bind=engine)
s = Session()

developers = s.query(Developer).all()
projects = s.query(Project).all()

for d in developers:
    log.debug(d)
    for p in d.projects:
        log.debug('    %s' % p)

for p in projects:
    log.debug(p)
    for d in p.developers:
        log.debug('    %s' % d)

# does not work
activeDevelopers = s.query(Developer).filter_by(Developer.developerProjects.status == 'active').all()
# AttributeError: Neither 'InstrumentedAttribute' object nor 'Comparator' object has an attribute 'status'

activeDevelopers = s.query(Developer).filter(Developer.developerProjects.any(status='active'))


#Example 2

character_relationships = Table('character_relationships', Base.metadata,
    Column('me_id', Integer, ForeignKey('characters.id', ondelete=True, onupdate=True), nullable=False, primary_key=True),
    Column('other_id', Integer, ForeignKey('characters.id', ondelete=True, onupdate=True), nullable=False, primary_key=True),
    UniqueConstraint('me_id', 'other_id', name='uix_1')
)

class CharacterRelationship(object):

    def __init__(self, me_id, other_id):
        self.me_id = me_id
        self.other_id = other_id

mapper(CharacterRelationship, character_relationships)

class Character(IdMixin, TimestampMixin, Base):
    __tablename__ = "characters"

    name = Column(Unicode, nullable=False)

    friends = relationship(lambda: Character, secondary=character_relationships, 
             primaryjoin=lambda: Character.id==character_relationships.c.me_id,
             secondaryjoin=lambda: Character.id==character_relationships.c.other_id,
             backref=backref('knows_me')
             )
             
@property
def real_friends(self):
    return set(my_character.friends) & set(my_character.knows_me)
    
@property
def real_friends(self):
    char_rels1 = aliased(CharacterRelationship)
    char_rels2 = aliased(CharacterRelationship)
    return DBSession.query(Character).\
        join(char_rels1, char_rels1.other_id == Character.id).\
        filter(char_rels1.me_id == self.id).\
        join(char_rels2, char_rels2.me_id == Character.id).\
        filter(char_rels2.other_id == self.id).all()
        
        
        
#Example 3
from sqlalchemy import Integer, Table, Column, ForeignKey, \
    create_engine, String, select
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base()

friendship = Table(
    'friendships', Base.metadata,
    Column('friend_a_id', Integer, ForeignKey('users.id'), 
                                        primary_key=True),
    Column('friend_b_id', Integer, ForeignKey('users.id'), 
                                        primary_key=True)
)


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    # this relationship is used for persistence
    friends = relationship("User", secondary=friendship, 
                           primaryjoin=id==friendship.c.friend_a_id,
                           secondaryjoin=id==friendship.c.friend_b_id,
    )

    def __repr__(self):
        return "User(%r)" % self.name

# this relationship is viewonly and selects across the union of all
# friends
friendship_union = select([
                        friendship.c.friend_a_id, 
                        friendship.c.friend_b_id
                        ]).union(
                            select([
                                friendship.c.friend_b_id, 
                                friendship.c.friend_a_id]
                            )
                    ).alias()
User.all_friends = relationship('User',
                       secondary=friendship_union,
                       primaryjoin=User.id==friendship_union.c.friend_a_id,
                       secondaryjoin=User.id==friendship_union.c.friend_b_id,
                       viewonly=True) 

e = create_engine("sqlite://",echo=True)
Base.metadata.create_all(e)
s = Session(e)

u1, u2, u3, u4, u5 = User(name='u1'), User(name='u2'), \
                    User(name='u3'), User(name='u4'), User(name='u5')

u1.friends = [u2, u3]
u4.friends = [u2, u5]
u3.friends.append(u5)
s.add_all([u1, u2, u3, u4, u5])
s.commit()

print u2.all_friends
print u5.all_friends

Is there any way of making the friendship table using an association object?