Connect Flask to PostPenguin

Flask is a lightweight Python web framework. This guide shows you how to create a webhook endpoint that receives and stores PostPenguin blog posts.

🚀 Quick Setup

1. Install Dependencies

pip install flask flask-sqlalchemy psycopg2-binary python-dotenv

2. Create the Flask App

# app.py
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import hmac
import hashlib
import os
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

# Post model
class Post(db.Model):
    __tablename__ = 'posts'
    
    id = db.Column(db.Integer, primary_key=True)
    postpenguin_id = db.Column(db.String(255), unique=True, nullable=False)
    title = db.Column(db.String(500), nullable=False)
    slug = db.Column(db.String(255), unique=True, nullable=False)
    html = db.Column(db.Text, nullable=False)
    meta_title = db.Column(db.String(500))
    meta_description = db.Column(db.Text)
    featured_image = db.Column(db.String(1000))
    tags = db.Column(db.JSON, default=[])
    status = db.Column(db.String(50), default='publish')
    published_at = db.Column(db.DateTime, default=datetime.utcnow)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    def to_dict(self):
        return {
            'id': self.id,
            'postPenguinId': self.postpenguin_id,
            'title': self.title,
            'slug': self.slug,
            'html': self.html,
            'metaTitle': self.meta_title,
            'metaDescription': self.meta_description,
            'featuredImage': self.featured_image,
            'tags': self.tags,
            'status': self.status,
            'publishedAt': self.published_at.isoformat() if self.published_at else None,
            'createdAt': self.created_at.isoformat() if self.created_at else None,
        }

def verify_signature(payload, signature, secret):
    """Verify HMAC-SHA256 webhook signature"""
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    received = signature.replace('sha256=', '')
    return hmac.compare_digest(expected, received)

@app.route('/api/webhooks/postpenguin', methods=['POST'])
def webhook():
    """Receive blog posts from PostPenguin"""
    try:
        # Verify signature
        signature = request.headers.get('X-PostPenguin-Signature', '')
        secret = os.getenv('POSTPENGUIN_WEBHOOK_SECRET', '')
        
        if secret and signature:
            if not verify_signature(request.get_data(as_text=True), signature, secret):
                return jsonify({'error': 'Invalid signature'}), 401
        
        data = request.get_json()
        
        # Validate required fields
        if not all(k in data for k in ['title', 'slug', 'html']):
            return jsonify({'error': 'Missing required fields'}), 400
        
        postpenguin_id = data.get('postPenguinId', f'pp_{int(datetime.now().timestamp())}')
        
        # Check if post exists
        existing_post = Post.query.filter_by(postpenguin_id=postpenguin_id).first()
        
        if existing_post:
            # Update existing post
            existing_post.title = data['title']
            existing_post.slug = data['slug']
            existing_post.html = data['html']
            existing_post.meta_title = data.get('meta_title', data['title'])
            existing_post.meta_description = data.get('meta_description', '')
            existing_post.featured_image = data.get('featured_image')
            existing_post.tags = data.get('tags', [])
            action = 'updated'
            post = existing_post
        else:
            # Create new post
            post = Post(
                postpenguin_id=postpenguin_id,
                title=data['title'],
                slug=data['slug'],
                html=data['html'],
                meta_title=data.get('meta_title', data['title']),
                meta_description=data.get('meta_description', ''),
                featured_image=data.get('featured_image'),
                tags=data.get('tags', []),
            )
            db.session.add(post)
            action = 'created'
        
        db.session.commit()
        
        return jsonify({
            'success': True,
            'postId': post.id,
            'postPenguinId': post.postpenguin_id,
            'action': action
        })
        
    except Exception as e:
        db.session.rollback()
        app.logger.error(f'Webhook error: {e}')
        return jsonify({'error': 'Internal server error'}), 500

@app.route('/api/posts', methods=['GET'])
def get_posts():
    """Fetch posts from database"""
    try:
        slug = request.args.get('slug')
        limit = int(request.args.get('limit', 10))
        offset = int(request.args.get('offset', 0))
        status = request.args.get('status', 'publish')
        
        if slug:
            post = Post.query.filter_by(slug=slug, status=status).first()
            if not post:
                return jsonify({'error': 'Post not found'}), 404
            return jsonify({'post': post.to_dict()})
        
        query = Post.query.filter_by(status=status)
        total = query.count()
        posts = query.order_by(Post.published_at.desc()).offset(offset).limit(limit).all()
        
        return jsonify({
            'posts': [p.to_dict() for p in posts],
            'pagination': {
                'total': total,
                'limit': limit,
                'offset': offset
            }
        })
        
    except Exception as e:
        app.logger.error(f'Error fetching posts: {e}')
        return jsonify({'error': 'Internal server error'}), 500

@app.route('/health', methods=['GET'])
def health():
    """Health check endpoint"""
    try:
        count = Post.query.count()
        return jsonify({'status': 'ok', 'posts_count': count})
    except:
        return jsonify({'status': 'error', 'database': 'disconnected'}), 500

# Create tables
with app.app_context():
    db.create_all()

if __name__ == '__main__':
    app.run(debug=True, port=3001)

⚙️ Environment Variables

# .env
DATABASE_URL=postgresql://user:password@localhost:5432/database
POSTPENGUIN_WEBHOOK_SECRET=your-secret-key-here
FLASK_ENV=development

🧪 Testing

Run the Server

# Development
python app.py

# Or with Flask CLI
flask run --port 3001

Test Webhook

curl -X POST http://localhost:3001/api/webhooks/postpenguin \
  -H "Content-Type: application/json" \
  -d '{
    "postPenguinId": "test_flask_123",
    "title": "Test Flask Post",
    "slug": "test-flask-post",
    "html": "<p className="text-gray-700">This is a test post using Flask.</p>",
    "meta_title": "Test Post",
    "meta_description": "Testing Flask webhook",
    "tags": ["test", "flask", "python"]
  }'

Test Posts API

# Get all posts
curl http://localhost:3001/api/posts

# Get single post
curl "http://localhost:3001/api/posts?slug=test-flask-post"

# Health check
curl http://localhost:3001/health

🚀 Production Deployment

Using Gunicorn

pip install gunicorn

gunicorn app:app -w 4 -b 0.0.0.0:3001

Docker

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

EXPOSE 3001
CMD ["gunicorn", "app:app", "-w", "4", "-b", "0.0.0.0:3001"]