Practical usage of SQLAlchemy – Blogging Application

This is a Python script that shows how a SQLite database can be created and maintained by a blogging application.
I’ve created a comprehensive blogging application that demonstrates SQLAlchemy in a real-world scenario. Here’s what the application includes:

🏗️ Database Models

• User: Authors and commenters with profiles
• Category: Post categories for organization
• Tag: Many-to-many relationship with posts
• Post: Blog posts with rich metadata
• Comment: User comments on posts

🔧 Key Features

Advanced Relationships

• One-to-many: User → Posts, Post → Comments
• Many-to-many: Posts ↔ Tags
• Foreign key constraints and cascading deletes

Database Operations Class

• Context manager for proper session handling
• CRUD operations for all entities
• Advanced queries (search, filtering, statistics)
• Slug generation for SEO-friendly URLs

Real-World Functionality

• Content Management: Create/publish posts, manage drafts
• Search: Full-text search across titles and content
• Analytics: Blog statistics and popular content
• Social Features: Comments and user interactions

Demo Features

• Sample Data: Realistic blog content with multiple authors
• Interactive CLI: Browse users, posts, categories, and search
• Statistics Dashboard: User counts, post metrics, popular tags
• Recent Activity: Latest posts and comments

📊 Advanced SQLAlchemy Concepts Demonstrated

1. Complex Relationships: Multiple relationship types with proper back-references
2. Association Tables: Many-to-many relationships with post_tags
3. Query Optimization: Efficient joins and filtering
4. Aggregate Functions: Counting, grouping, and statistics
5. Search Functionality: LIKE queries and content filtering
6. Date Handling: Timestamps and date-based queries
7. Session Management: Proper transaction handling and cleanup

🚀 How to Run

Simply execute the script and it will:

1. Create the SQLite database and tables
2. Generate sample data (users, posts, comments, tags)
3. Display blog statistics and recent content
4. Launch an interactive menu for exploring the data

The application creates a real SQLite database file (blog_app.db) that you can inspect with any SQLite browser to see the actual database structure and data.

This example showcases how SQLAlchemy can handle complex real-world scenarios while keeping the code clean and maintainable!

"""
Complete Blogging Application with SQLAlchemy
=============================================

This is a comprehensive example of a blogging application using SQLAlchemy.
It includes models, database operations, and a simple CLI interface.

Run this script to see the blogging application in action!
"""

import os
from datetime import datetime, timedelta
from typing import List, Optional
from sqlalchemy import (
    create_engine, Column, Integer, String, Text, DateTime, 
    Boolean, ForeignKey, Table, func, desc
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, Session

# Database setup
DATABASE_URL = "sqlite:///blog_app.db"
engine = create_engine(DATABASE_URL, echo=False)  # Set to True to see SQL queries
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Association table for many-to-many relationship between posts and tags
post_tags = Table(
    'post_tags',
    Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

# Models
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False, index=True)
    full_name = Column(String(100), nullable=False)
    bio = Column(Text)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # Relationships
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")
    comments = relationship("Comment", back_populates="author", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class Category(Base):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False, index=True)
    description = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    posts = relationship("Post", back_populates="category")
    
    def __repr__(self):
        return f"<Category(name='{self.name}')>"

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(30), unique=True, nullable=False, index=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    posts = relationship("Post", secondary=post_tags, back_populates="tags")
    
    def __repr__(self):
        return f"<Tag(name='{self.name}')>"

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False, index=True)
    slug = Column(String(250), unique=True, nullable=False, index=True)
    content = Column(Text, nullable=False)
    excerpt = Column(Text)
    is_published = Column(Boolean, default=False)
    published_at = Column(DateTime)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # Foreign keys
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    category_id = Column(Integer, ForeignKey('categories.id'))
    
    # Relationships
    author = relationship("User", back_populates="posts")
    category = relationship("Category", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")
    tags = relationship("Tag", secondary=post_tags, back_populates="posts")
    
    def __repr__(self):
        return f"<Post(title='{self.title}', author='{self.author.username if self.author else None}')>"

class Comment(Base):
    __tablename__ = 'comments'
    
    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)
    is_approved = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Foreign keys
    post_id = Column(Integer, ForeignKey('posts.id'), nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    
    # Relationships
    post = relationship("Post", back_populates="comments")
    author = relationship("User", back_populates="comments")
    
    def __repr__(self):
        return f"<Comment(post='{self.post.title if self.post else None}', author='{self.author.username if self.author else None}')>"

# Database operations
class BlogDatabase:
    def __init__(self):
        self.session = SessionLocal()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.session.rollback()
        self.session.close()
    
    def create_user(self, username: str, email: str, full_name: str, bio: str = None) -> User:
        """Create a new user"""
        user = User(
            username=username,
            email=email,
            full_name=full_name,
            bio=bio
        )
        self.session.add(user)
        self.session.commit()
        self.session.refresh(user)
        return user
    
    def create_category(self, name: str, description: str = None) -> Category:
        """Create a new category"""
        category = Category(name=name, description=description)
        self.session.add(category)
        self.session.commit()
        self.session.refresh(category)
        return category
    
    def create_tag(self, name: str) -> Tag:
        """Create a new tag or return existing one"""
        tag = self.session.query(Tag).filter(Tag.name == name).first()
        if not tag:
            tag = Tag(name=name)
            self.session.add(tag)
            self.session.commit()
            self.session.refresh(tag)
        return tag
    
    def create_post(self, title: str, content: str, author_id: int, 
                   category_id: int = None, tags: List[str] = None, 
                   is_published: bool = False) -> Post:
        """Create a new blog post"""
        # Generate slug from title
        slug = title.lower().replace(' ', '-').replace('.', '').replace(',', '')
        
        post = Post(
            title=title,
            slug=slug,
            content=content,
            excerpt=content[:200] + "..." if len(content) > 200 else content,
            author_id=author_id,
            category_id=category_id,
            is_published=is_published,
            published_at=datetime.utcnow() if is_published else None
        )
        
        # Add tags if provided
        if tags:
            for tag_name in tags:
                tag = self.create_tag(tag_name)
                post.tags.append(tag)
        
        self.session.add(post)
        self.session.commit()
        self.session.refresh(post)
        return post
    
    def create_comment(self, content: str, post_id: int, author_id: int) -> Comment:
        """Create a new comment"""
        comment = Comment(
            content=content,
            post_id=post_id,
            author_id=author_id,
            is_approved=True  # Auto-approve for demo
        )
        self.session.add(comment)
        self.session.commit()
        self.session.refresh(comment)
        return comment
    
    def get_user_by_username(self, username: str) -> Optional[User]:
        """Get user by username"""
        return self.session.query(User).filter(User.username == username).first()
    
    def get_all_users(self) -> List[User]:
        """Get all users"""
        return self.session.query(User).all()
    
    def get_published_posts(self, limit: int = 10) -> List[Post]:
        """Get published posts ordered by published date"""
        return (self.session.query(Post)
                .filter(Post.is_published == True)
                .order_by(desc(Post.published_at))
                .limit(limit)
                .all())
    
    def get_posts_by_author(self, author_id: int) -> List[Post]:
        """Get all posts by a specific author"""
        return (self.session.query(Post)
                .filter(Post.author_id == author_id)
                .order_by(desc(Post.created_at))
                .all())
    
    def get_posts_by_category(self, category_id: int) -> List[Post]:
        """Get all posts in a specific category"""
        return (self.session.query(Post)
                .filter(Post.category_id == category_id)
                .filter(Post.is_published == True)
                .order_by(desc(Post.published_at))
                .all())
    
    def get_posts_by_tag(self, tag_name: str) -> List[Post]:
        """Get all posts with a specific tag"""
        return (self.session.query(Post)
                .join(Post.tags)
                .filter(Tag.name == tag_name)
                .filter(Post.is_published == True)
                .order_by(desc(Post.published_at))
                .all())
    
    def get_post_by_slug(self, slug: str) -> Optional[Post]:
        """Get a post by its slug"""
        return self.session.query(Post).filter(Post.slug == slug).first()
    
    def get_categories(self) -> List[Category]:
        """Get all categories"""
        return self.session.query(Category).all()
    
    def get_popular_tags(self, limit: int = 10) -> List[tuple]:
        """Get most popular tags with post counts"""
        return (self.session.query(Tag.name, func.count(post_tags.c.post_id).label('post_count'))
                .join(post_tags)
                .group_by(Tag.name)
                .order_by(desc('post_count'))
                .limit(limit)
                .all())
    
    def search_posts(self, query: str) -> List[Post]:
        """Search posts by title or content"""
        search_term = f"%{query}%"
        return (self.session.query(Post)
                .filter(
                    (Post.title.like(search_term)) | 
                    (Post.content.like(search_term))
                )
                .filter(Post.is_published == True)
                .order_by(desc(Post.published_at))
                .all())
    
    def get_recent_comments(self, limit: int = 10) -> List[Comment]:
        """Get recent approved comments"""
        return (self.session.query(Comment)
                .filter(Comment.is_approved == True)
                .order_by(desc(Comment.created_at))
                .limit(limit)
                .all())
    
    def get_blog_stats(self) -> dict:
        """Get blog statistics"""
        return {
            'total_users': self.session.query(User).count(),
            'total_posts': self.session.query(Post).count(),
            'published_posts': self.session.query(Post).filter(Post.is_published == True).count(),
            'total_comments': self.session.query(Comment).count(),
            'total_categories': self.session.query(Category).count(),
            'total_tags': self.session.query(Tag).count()
        }

# Sample data and demo functions
def create_sample_data():
    """Create sample data for the blog"""
    print("Creating sample data...")
    
    with BlogDatabase() as db:
        # Create users
        alice = db.create_user(
            username="alice_writer",
            email="alice@blog.com",
            full_name="Alice Johnson",
            bio="Tech writer and Python enthusiast"
        )
        
        bob = db.create_user(
            username="bob_coder",
            email="bob@blog.com",
            full_name="Bob Smith",
            bio="Full-stack developer and blogger"
        )
        
        # Create categories
        tech_cat = db.create_category(
            name="Technology",
            description="Posts about technology and programming"
        )
        
        tutorial_cat = db.create_category(
            name="Tutorials",
            description="Step-by-step tutorials and guides"
        )
        
        # Create posts
        post1 = db.create_post(
            title="Getting Started with SQLAlchemy",
            content="""SQLAlchemy is a powerful Python SQL toolkit and Object-Relational Mapping (ORM) library. 
            In this post, we'll explore the basics of SQLAlchemy and how to use it in your Python applications.
            
            SQLAlchemy provides a full suite of well known enterprise-level persistence patterns, 
            designed for efficient and high-performing database access.""",
            author_id=alice.id,
            category_id=tech_cat.id,
            tags=["python", "sqlalchemy", "database", "orm"],
            is_published=True
        )
        
        post2 = db.create_post(
            title="Building REST APIs with FastAPI",
            content="""FastAPI is a modern, fast web framework for building APIs with Python 3.7+.
            It's built on standard Python type hints and provides automatic API documentation.
            
            In this tutorial, we'll build a complete REST API from scratch using FastAPI and SQLAlchemy.""",
            author_id=bob.id,
            category_id=tutorial_cat.id,
            tags=["python", "fastapi", "api", "web development"],
            is_published=True
        )
        
        post3 = db.create_post(
            title="Database Design Best Practices",
            content="""Good database design is crucial for application performance and maintainability.
            Here are some key principles to follow when designing your database schema.
            
            We'll cover normalization, indexing, relationships, and common pitfalls to avoid.""",
            author_id=alice.id,
            category_id=tech_cat.id,
            tags=["database", "design", "best practices"],
            is_published=True
        )
        
        # Create comments
        db.create_comment(
            content="Great introduction to SQLAlchemy! Very helpful for beginners.",
            post_id=post1.id,
            author_id=bob.id
        )
        
        db.create_comment(
            content="I've been using FastAPI for a while now, and this tutorial covers all the essentials.",
            post_id=post2.id,
            author_id=alice.id
        )
        
        db.create_comment(
            content="Database design is often overlooked. Thanks for sharing these tips!",
            post_id=post3.id,
            author_id=bob.id
        )
    
    print("Sample data created successfully!")

def display_blog_stats():
    """Display blog statistics"""
    with BlogDatabase() as db:
        stats = db.get_blog_stats()
        print("\n📊 Blog Statistics:")
        print("=" * 30)
        for key, value in stats.items():
            print(f"{key.replace('_', ' ').title()}: {value}")

def display_recent_posts():
    """Display recent published posts"""
    with BlogDatabase() as db:
        posts = db.get_published_posts(limit=5)
        print("\n📝 Recent Posts:")
        print("=" * 40)
        for post in posts:
            print(f"📄 {post.title}")
            print(f"   By: {post.author.username} | Category: {post.category.name if post.category else 'Uncategorized'}")
            print(f"   Published: {post.published_at.strftime('%Y-%m-%d %H:%M')}")
            print(f"   Tags: {', '.join([tag.name for tag in post.tags])}")
            print(f"   {post.excerpt}")
            print("-" * 40)

def display_popular_tags():
    """Display popular tags"""
    with BlogDatabase() as db:
        tags = db.get_popular_tags(limit=5)
        print("\n🏷️  Popular Tags:")
        print("=" * 25)
        for tag_name, count in tags:
            print(f"#{tag_name} ({count} posts)")

def display_recent_comments():
    """Display recent comments"""
    with BlogDatabase() as db:
        comments = db.get_recent_comments(limit=5)
        print("\n💬 Recent Comments:")
        print("=" * 30)
        for comment in comments:
            print(f"🗨️  {comment.author.username} on '{comment.post.title}':")
            print(f"   {comment.content[:100]}{'...' if len(comment.content) > 100 else ''}")
            print(f"   {comment.created_at.strftime('%Y-%m-%d %H:%M')}")
            print("-" * 30)

def search_demo():
    """Demonstrate search functionality"""
    with BlogDatabase() as db:
        search_query = "SQLAlchemy"
        results = db.search_posts(search_query)
        print(f"\n🔍 Search Results for '{search_query}':")
        print("=" * 35)
        if results:
            for post in results:
                print(f"📄 {post.title}")
                print(f"   By: {post.author.username}")
                print(f"   {post.excerpt}")
                print("-" * 35)
        else:
            print("No results found.")

def main():
    """Main demo function"""
    print("🚀 Blogging Application Demo")
    print("=" * 50)
    
    # Create tables
    Base.metadata.create_all(bind=engine)
    
    # Check if sample data already exists
    with BlogDatabase() as db:
        if db.session.query(User).count() == 0:
            create_sample_data()
    
    # Display various information
    display_blog_stats()
    display_recent_posts()
    display_popular_tags()
    display_recent_comments()
    search_demo()
    
    print("\n✅ Demo completed! Check the 'blog_app.db' file for the SQLite database.")
    
    # Interactive menu
    while True:
        print("\n" + "=" * 50)
        print("Interactive Menu:")
        print("1. View all users")
        print("2. View posts by author")
        print("3. View posts by category")
        print("4. Search posts")
        print("5. Exit")
        
        choice = input("\nEnter your choice (1-5): ").strip()
        
        if choice == '1':
            with BlogDatabase() as db:
                users = db.get_all_users()
                print("\n👥 All Users:")
                for user in users:
                    print(f"  • {user.full_name} (@{user.username}) - {user.email}")
        
        elif choice == '2':
            username = input("Enter username: ").strip()
            with BlogDatabase() as db:
                user = db.get_user_by_username(username)
                if user:
                    posts = db.get_posts_by_author(user.id)
                    print(f"\n📝 Posts by {user.full_name}:")
                    for post in posts:
                        status = "✅ Published" if post.is_published else "📝 Draft"
                        print(f"  • {post.title} ({status})")
                else:
                    print("User not found.")
        
        elif choice == '3':
            with BlogDatabase() as db:
                categories = db.get_categories()
                print("\n📂 Categories:")
                for i, cat in enumerate(categories, 1):
                    print(f"  {i}. {cat.name}")
                
                try:
                    cat_choice = int(input("Enter category number: ")) - 1
                    if 0 <= cat_choice < len(categories):
                        posts = db.get_posts_by_category(categories[cat_choice].id)
                        print(f"\n📝 Posts in '{categories[cat_choice].name}':")
                        for post in posts:
                            print(f"  • {post.title} by {post.author.username}")
                    else:
                        print("Invalid category number.")
                except ValueError:
                    print("Please enter a valid number.")
        
        elif choice == '4':
            query = input("Enter search query: ").strip()
            if query:
                with BlogDatabase() as db:
                    results = db.search_posts(query)
                    print(f"\n🔍 Search Results for '{query}':")
                    if results:
                        for post in results:
                            print(f"  • {post.title} by {post.author.username}")
                    else:
                        print("No results found.")
        
        elif choice == '5':
            print("👋 Goodbye!")
            break
        
        else:
            print("Invalid choice. Please try again.")

if __name__ == "__main__":
    main()