Connect FastAPI to PostPenguin

FastAPI is a modern, high-performance Python framework. This guide shows you how to create a webhook endpoint with automatic validation and async database support.

🚀 Quick Setup

1. Install Dependencies

pip install fastapi uvicorn sqlalchemy asyncpg python-dotenv

2. Create the Webhook Handler

# main.py
from fastapi import FastAPI, HTTPException, Header, Request
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
import hmac
import hashlib
import os
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="PostPenguin Webhook Handler")

# Pydantic models for request validation
class PostPenguinPayload(BaseModel):
    postPenguinId: Optional[str] = None
    title: str
    slug: str
    html: str
    meta_title: Optional[str] = None
    meta_description: Optional[str] = None
    featured_image: Optional[str] = None
    tags: Optional[List[str]] = []
    publishedAt: Optional[str] = None

class WebhookResponse(BaseModel):
    success: bool
    postId: str
    message: str

# In-memory storage for demo (replace with database)
posts_db = {}

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

@app.post("/api/webhooks/postpenguin", response_model=WebhookResponse)
async def receive_webhook(
    request: Request,
    payload: PostPenguinPayload,
    x_postpenguin_signature: Optional[str] = Header(None)
):
    """
    Receive blog posts from PostPenguin.
    
    This endpoint:
    - Verifies the webhook signature
    - Validates the payload
    - Saves the post to your database
    """
    # Verify signature if secret is configured
    secret = os.getenv('POSTPENGUIN_WEBHOOK_SECRET')
    
    if secret and x_postpenguin_signature:
        body = await request.body()
        if not verify_signature(body, x_postpenguin_signature, secret):
            raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Generate ID if not provided
    post_id = payload.postPenguinId or f"pp_{int(datetime.now().timestamp())}"
    
    # Create post object
    post = {
        "id": post_id,
        "title": payload.title,
        "slug": payload.slug,
        "html": payload.html,
        "meta_title": payload.meta_title or payload.title,
        "meta_description": payload.meta_description or "",
        "featured_image": payload.featured_image,
        "tags": payload.tags,
        "published_at": payload.publishedAt or datetime.now().isoformat(),
        "created_at": datetime.now().isoformat(),
        "updated_at": datetime.now().isoformat(),
    }
    
    # Save to database (demo: in-memory)
    action = "updated" if post_id in posts_db else "created"
    posts_db[post_id] = post
    
    return WebhookResponse(
        success=True,
        postId=post_id,
        message=f"Post {action} successfully"
    )

@app.get("/api/posts")
async def get_posts(
    slug: Optional[str] = None,
    limit: int = 10,
    offset: int = 0
):
    """Fetch posts from the database"""
    posts = list(posts_db.values())
    
    if slug:
        post = next((p for p in posts if p["slug"] == slug), None)
        if not post:
            raise HTTPException(status_code=404, detail="Post not found")
        return {"post": post}
    
    # Sort by published_at descending
    posts.sort(key=lambda x: x["published_at"], reverse=True)
    
    return {
        "posts": posts[offset:offset + limit],
        "pagination": {
            "total": len(posts),
            "limit": limit,
            "offset": offset
        }
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "ok", "posts_count": len(posts_db)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=3001)

💾 With SQLAlchemy + PostgreSQL

# models.py
from sqlalchemy import Column, String, Text, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

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

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

# main.py with database
from fastapi import FastAPI, HTTPException, Header, Request, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from database import get_db
from models import Post
import uuid

@app.post("/api/webhooks/postpenguin")
async def receive_webhook(
    request: Request,
    payload: PostPenguinPayload,
    x_postpenguin_signature: Optional[str] = Header(None),
    db: AsyncSession = Depends(get_db)
):
    # Verify signature...
    
    post_id = payload.postPenguinId or f"pp_{uuid.uuid4().hex[:8]}"
    
    # Check if post exists
    result = await db.execute(
        select(Post).where(Post.postpenguin_id == post_id)
    )
    existing_post = result.scalar_one_or_none()
    
    if existing_post:
        # Update existing post
        existing_post.title = payload.title
        existing_post.slug = payload.slug
        existing_post.html = payload.html
        existing_post.meta_title = payload.meta_title or payload.title
        existing_post.meta_description = payload.meta_description or ""
        existing_post.featured_image = payload.featured_image
        existing_post.tags = payload.tags
        action = "updated"
    else:
        # Create new post
        new_post = Post(
            id=str(uuid.uuid4()),
            postpenguin_id=post_id,
            title=payload.title,
            slug=payload.slug,
            html=payload.html,
            meta_title=payload.meta_title or payload.title,
            meta_description=payload.meta_description or "",
            featured_image=payload.featured_image,
            tags=payload.tags,
        )
        db.add(new_post)
        action = "created"
    
    await db.commit()
    
    return {"success": True, "postId": post_id, "action": action}

⚙️ Environment Variables

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

🧪 Testing

Run the Server

uvicorn main:app --reload --port 3001

View Auto-Generated Docs

FastAPI automatically generates interactive API documentation:

• Swagger UI: http://localhost:3001/docs
• ReDoc: http://localhost:3001/redoc

Test Webhook

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

🚀 Production Deployment

Using Gunicorn + Uvicorn

pip install gunicorn

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -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 ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "3001"]