Quick Start

Build a complete blog API with Zenith in just 5 minutes. This tutorial shows you Zenith’s core features and how they work together.

What you’ll learn:

Creating a Zenith application with automatic configuration
Using ZenithModel for intuitive database operations
Adding authentication, admin panels, and documentation with one-liners
Building a complete CRUD API with minimal code

Installation

pip install zenithweb

Or with uv (recommended):

uv add zenithweb

Create a New Project

Use the Zenith CLI to create a new project with all the essentials:

zen new blog-api
cd blog-api

This creates:

app.py - Main application with sample endpoints
.env - Environment variables with generated secret key
requirements.txt - Dependencies
.gitignore - Git ignore rules
README.md - Quick start guide

Building a Blog API

Create a file called blog.py:

from zenith import Zenith
from zenith.db import ZenithModel
from zenith import Auth
from sqlmodel import Field
from datetime import datetime
from typing import Optional

# Create application with automatic configuration
app = Zenith()

# Add built-in features with one-liners
app.add_auth()                      # JWT authentication system
app.add_admin("/admin")             # Admin dashboard
app.add_api("Blog API", "1.0.0")    # Adds /docs and /redoc endpoints

# Define blog post model
class Post(ZenithModel, table=True):
    """Blog post with automatic session management."""
    id: Optional[int] = Field(primary_key=True)
    title: str = Field(index=True)
    content: str
    published: bool = Field(default=False)
    author_id: int
    created_at: datetime = Field(default_factory=datetime.now)

# User model for authentication
class User(ZenithModel, table=True):
    """User model for authentication."""
    id: Optional[int] = Field(primary_key=True)
    email: str = Field(unique=True, index=True)
    name: str
    hashed_password: str

# Blog endpoints with automatic session management
@app.post("/posts")
async def create_post(title: str, content: str, user=Auth):
    """Create a new blog post."""
    post = await Post.create(
        title=title,
        content=content,
        author_id=user.id
    )
    return {"message": "Post created", "post": post.model_dump()}

@app.get("/posts")
async def list_posts(
    published: Optional[bool] = None,
    limit: int = 10,
    offset: int = 0
):
    """List blog posts with optional filtering."""
    # Build query based on filters
    if published is not None:
        query = Post.where(published=published)
    else:
        query = Post.where()  # All posts

    posts = await query.order_by('-created_at').limit(limit).offset(offset).all()

    return {
        "posts": [p.model_dump() for p in posts],
        "total": await Post.count(),  # Use class method for total count
        "limit": limit,
        "offset": offset
    }

@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    """Get a single post by ID."""
    post = await Post.find_or_404(post_id)
    return {"post": post.model_dump()}

@app.patch("/posts/{post_id}")
async def update_post(post_id: int, title: str = None, content: str = None, user=Auth):
    """Update a blog post."""
    post = await Post.find_or_404(post_id)

    # Simple authorization check
    if post.author_id != user.id:
        raise HTTPException(403, "Not authorized to edit this post")

    updates = {}
    if title is not None:
        updates["title"] = title
    if content is not None:
        updates["content"] = content

    if updates:
        await post.update(**updates)

    return {"message": "Post updated", "post": post.model_dump()}

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, user=Auth):
    """Delete a blog post."""
    post = await Post.find_or_404(post_id)

    if post.author_id != user.id:
        raise HTTPException(403, "Not authorized to delete this post")

    await post.delete()
    return {"message": "Post deleted"}

@app.patch("/posts/{post_id}/publish")
async def publish_post(post_id: int, user=Auth):
    """Publish a blog post."""
    post = await Post.find_or_404(post_id)

    if post.author_id != user.id:
        raise HTTPException(403, "Not authorized")

    await post.update(published=True)
    return {"message": "Post published", "post": post.model_dump()}

# Optional: Add custom endpoint for user's posts
@app.get("/my-posts")
async def my_posts(user=Auth):
    """Get current user's posts."""
    posts = await Post.where(author_id=user.id).order_by('-created_at').all()
    return {"posts": [p.model_dump() for p in posts]}

Run Your API

Start the development server:

# Using uvicorn directly
uvicorn blog:app --reload

# Or using Zenith CLI
zen dev blog.py

Visit http://localhost:8000 and you’ll see your API running!

What You Get

Your blog API now includes:

Automatic Features

Database: SQLite for development (configures PostgreSQL for production)
Migrations: Automatic table creation and schema updates
Session management: Request-scoped database sessions
Type safety: Full validation with Pydantic models
Error handling: Structured error responses

Authentication System (`/auth/login`)

# Create a user account (you'll need to add a registration endpoint)
curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "demo@example.com", "password": "password"}'

Admin Dashboard (`/admin`)

System health monitoring
Database statistics
Application metrics
Request logs

Interactive Documentation (`/docs`)

Try all endpoints directly in your browser
Automatic request/response examples
Authentication testing
OpenAPI schema export

API Endpoints

# List posts
curl http://localhost:8000/posts

# Get specific post
curl http://localhost:8000/posts/1

# Create post (requires authentication)
curl -X POST http://localhost:8000/posts \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"title": "My First Post", "content": "Hello, World!"}'

# Publish post
curl -X PATCH http://localhost:8000/posts/1/publish \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

ZenithModel Features

The ZenithModel class provides Rails-like query methods:

# Finding records
post = await Post.find(1)              # Returns None if not found
post = await Post.find_or_404(1)       # Raises 404 if not found
posts = await Post.all()               # Get all records

# Querying with conditions
published = await Post.where(published=True).all()
recent = await Post.where(published=True).order_by('-created_at').limit(5).all()

# Creating records
post = await Post.create(title="New Post", content="Content", author_id=1)

# Updating records
await post.update(published=True)

# Deleting records
await post.delete()

# Counting
count = await Post.where(published=True).count()

# Serialization
data = post.model_dump()                   # Convert to dictionary
post = Post(**data)                     # Create from dictionary

Next Steps

Add User Registration

from zenith.auth.password import hash_password

@app.post("/register")
async def register(email: str, name: str, password: str):
    """Register a new user."""
    # Check if user exists
    existing = await User.where(email=email).first()
    if existing:
        raise HTTPException(400, "Email already registered")

    # Create user
    user = await User.create(
        email=email,
        name=name,
        hashed_password=hash_password(password)
    )

    return {"message": "User registered", "user_id": user.id}

Add Pagination

@app.get("/posts")
async def list_posts(page: int = 1, per_page: int = 10):
    """List posts with pagination."""
    offset = (page - 1) * per_page

    posts = await Post.where(published=True).limit(per_page).offset(offset).all()
    total = await Post.where(published=True).count()

    return {
        "posts": [p.model_dump() for p in posts],
        "page": page,
        "per_page": per_page,
        "total": total,
        "pages": (total + per_page - 1) // per_page
    }

Add Search

from sqlalchemy import or_

@app.get("/search")
async def search_posts(q: str):
    """Search posts by title or content."""
    session = await Post._get_session()

    stmt = select(Post).where(
        or_(
            Post.title.ilike(f"%{q}%"),
            Post.content.ilike(f"%{q}%")
        )
    )

    result = await session.execute(stmt)
    posts = result.scalars().all()

    return {"query": q, "posts": [p.model_dump() for p in posts]}

Key Benefits

Development Speed

Start building immediately with zero configuration
Built-in authentication saves hours of setup
Auto-generated admin interface
Interactive documentation

Code Quality

Type safety throughout with full IDE support
Automatic validation prevents common bugs
Clean, readable query methods
Consistent error handling

Production Ready

Environment-aware configuration
Security middleware included
Performance monitoring built-in
Scalable async architecture

Learn More

Tutorial - Step-by-step guide building a complete application
Database Guide - Advanced ZenithModel patterns
Authentication - JWT setup and user management
API Reference - Complete framework documentation