test-orm-join.py

from sqlalchemy import (create_engine, Column, Integer, String, ForeignKey,
                        UniqueConstraint)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


Base = declarative_base()


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False, unique=True)
    fullname = Column(String(30))
    password = Column(String(30))


class Group(Base):
    __tablename__ = 'groups'
    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=False, unique=True)


class UserGroup(Base):
    __tablename__ = 'user_groups'
    id = Column(Integer, primary_key=True)
    user = Column(Integer, ForeignKey('users.id'), nullable=False)
    user_group = Column(Integer, ForeignKey('groups.id'), nullable=False)
    __table_args__ = (UniqueConstraint('user', 'user_group'),)


db_url = 'postgresql://sugiana:secret@localhost/test'
engine = create_engine(db_url)
engine.echo = True
Base.metadata.create_all(engine)
BaseSession = sessionmaker(bind=engine)
DBSession = BaseSession()

user_data = [
    'sugiana',
    ]

group_data = [
    'Administrator',
    ]

for user_name in user_data:
    user = DBSession.query(User).filter_by(name=user_name).first()
    if not user:
        user = User(name=user_name)
        DBSession.add(user)

for group_name in group_data:
    group = DBSession.query(Group).filter_by(name=group_name).first()
    if not group:
        group = Group(name=group_name)
        DBSession.add(group)

DBSession.flush() # send to database server agar mendapatkan ID user & group

ug = DBSession.query(UserGroup).filter_by(user=user.id, user_group=group.id).first()
if not ug:
    ug = UserGroup(user=user.id, user_group=group.id)
    DBSession.add(ug)

DBSession.commit()

query_ug = DBSession.query(UserGroup.id, User.name.label('user_name'),
            Group.name.label('group_name')).\
            filter(UserGroup.user==User.id).\
            filter(UserGroup.user_group==Group.id)
for ug in query_ug.order_by(User.name, Group.name):
    print('User: %s, Group: %s' % (ug.user_name, ug.group_name))