Skip to content

Instantly share code, notes, and snippets.

@gamesbrainiac
Created November 1, 2014 05:56
Show Gist options
  • Select an option

  • Save gamesbrainiac/cafa32d4c6f6b201ccd5 to your computer and use it in GitHub Desktop.

Select an option

Save gamesbrainiac/cafa32d4c6f6b201ccd5 to your computer and use it in GitHub Desktop.
An example based tutorial on how to create different types of relationships in SQLAlchemy
from datetime import datetime
import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///test.db')
Base = declarative_base(bind=engine)
Session = orm.sessionmaker(bind=engine)
# @event.listens_for(Session, 'before_commit')
# def update_all(session):
# for obj in [o for o in session if hasattr(o, 'updated_at') and o not in session.new]:
# obj.updated_at = datetime.utcnow()
class TimeStampMixin(object):
# This object demonstrates how to make a simple mixin in SQLAlchemy
created_at = sa.Column(sa.DateTime, default=datetime.utcnow())
updated_at = sa.Column(sa.DateTime, default=datetime.utcnow())
# Updates values every time a new value is set.
def __setattr__(self, key, value):
super().__setattr__(key, value)
super().__setattr__('updated_at', datetime.utcnow())
class User(Base, TimeStampMixin):
# This User table demonstrates how to create a simple table in SQLAlchemy.
# We inherit from base, and from a mixin
__tablename__ = 'users'
id = sa.Column(sa.Integer, primary_key=True)
username = sa.Column(sa.String, unique=True)
password = sa.Column(sa.String)
def __init__(self, username, password):
self.username = username
self.password = password
def __repr__(self):
return "[{}] {}".format(self.id, self.username)
class BlogPost(Base, TimeStampMixin):
# Demonstrates o2m
__tablename__ = 'blogposts'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(128), unique=True)
content = sa.Column(sa.String)
user_id = sa.Column(sa.ForeignKey("users.id"))
user = orm.relationship(User, backref=orm.backref('posts', order_by=id, cascade='all, delete, delete-orphan'))
def __init__(self, name, content, user):
self.name = name
self.user = user
self.content = content
def __repr__(self):
return "[{}] {} by <{}>".format(self.id, self.name, self.user)
class Tag(Base):
# Demonstrates m2m
__tablename__ = 'tags'
id = sa.Column('id', sa.Integer, primary_key=True)
name = sa.Column('name', sa.String(100), unique=True)
description = sa.Column('description', sa.String)
# Use lambdas when defining secondary tables, since they are easy to refactor
posts = orm.relationship(BlogPost, lambda: post_tag, backref=orm.backref('tags', order_by=id))
def __repr__(self):
return "Tag {}".format(self.name)
def __init__(self, name, description):
self.name = name
self.description = description
post_tag = sa.Table('post_tag', Base.metadata,
sa.Column('post_id', sa.Integer, sa.ForeignKey('{}.id'.format(BlogPost.__tablename__))),
sa.Column('tag_id', sa.Integer, sa.ForeignKey('{}.id'.format(Tag.__tablename__))))
class Comment(Base, TimeStampMixin):
__tablename__ = 'comments'
# Comment specific information
id = sa.Column(sa.Integer, primary_key=True)
content = sa.Column(sa.String)
# Relationship with blogpost
post_id = sa.Column(sa.ForeignKey('blogposts.id'))
post = orm.relationship(BlogPost, backref=orm.backref('comments', order_by=id, cascade='all, delete, '
'delete-orphan'))
# Relationship with comment replies
parent_id = sa.Column(sa.ForeignKey('comments.id'))
# This is the creation of an adjacency list. Ask me more if you want to understand this part
replies = orm.relationship("Comment", backref=orm.backref('parent', remote_side=[id]))
# Relationship with user
user_id = sa.Column(sa.ForeignKey('users.id'))
user = orm.relationship(User, backref=orm.backref('comments',
order_by='Comment.id', cascade='all, delete, delete-orphan'))
def __init__(self, content, user, post, comment=None):
self.content = content
self.user = user
self.post = post
if comment is not None:
self.parent = comment
def __repr__(self):
return "[{}] by <{}> on <{}>".format(self.id, self.user, self.post)
Base.metadata.create_all(bind=engine)
if __name__ == '__main__':
sess = Session()
u = User('John Doe', 'Jane is my love')
uu = User('Jane Doe', 'John is my love')
p = BlogPost('Hello World', 'This is the first time I\'m seeing this place', u)
c = Comment('This post sucks', u, p)
r = Comment('Oh yea? Your comment sucks', u, p, c)
rr = Comment('No, this post really is bad', uu, p, r)
t = Tag('Python', 'A programming lang created in 1991')
p.tags.append(t)
sess.add_all([u, p, c, r, t, r])
sess.commit()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment