Flask-SQLAlchemy User and Post Management with SQLAlchemy ORM

This code demonstrates how to use Flask-SQLAlchemy to create a simple user and post model, as well as related database operations such as creating a session, adding users and posts, retrieving users by username, retrieving posts by user, updating usernames, and deleting users and posts.

Technology Stack : Flask-SQLAlchemy, SQLAlchemy, SQLite

Code Type : The type of code

Code Difficulty : Intermediate

                
                    
from flask_sqlalchemy import SQLAlchemy, SQLAlchemyEventDispatcher
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, nullable=False)
    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    body = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'))
    author = relationship('User', back_populates='posts')

def create_user_session(engine):
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

def add_user(session, username):
    new_user = User(username=username)
    session.add(new_user)
    session.commit()

def add_post(session, title, body, author_id):
    new_post = Post(title=title, body=body, author_id=author_id)
    session.add(new_post)
    session.commit()

def get_user_by_username(session, username):
    return session.query(User).filter_by(username=username).first()

def get_posts_by_user(session, user_id):
    return session.query(Post).filter_by(author_id=user_id).all()

def update_user_username(session, user_id, new_username):
    user = session.query(User).get(user_id)
    if user:
        user.username = new_username
        session.commit()

def delete_user(session, user_id):
    user = session.query(User).get(user_id)
    if user:
        session.delete(user)
        session.commit()

def delete_post(session, post_id):
    post = session.query(Post).get(post_id)
    if post:
        session.delete(post)
        session.commit()

def create_engine_with_database_url():
    return create_engine('sqlite:///example.db')

# Usage of the functions
if __name__ == "__main__":
    engine = create_engine_with_database_url()
    Base.metadata.create_all(engine)
    session = create_user_session(engine)
    add_user(session, 'john_doe')
    add_post(session, 'Hello World', 'This is a test post.', 1)
    user = get_user_by_username(session, 'john_doe')
    posts = get_posts_by_user(session, 1)
    update_user_username(session, 1, 'johnny_doe')
    delete_user(session, 1)
    delete_post(session, 1)