Created
November 1, 2014 05:56
-
-
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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