Continue to Part 2
Ready to add a database? Continue to Data Models →
Welcome to the Zenith tutorial! Over the next seven parts, we’ll build TaskFlow - a complete task management API that demonstrates Zenith’s capabilities.
What you’ll learn:
We’re creating TaskFlow - a task management API similar to Todoist or Asana. Users can:
Before we begin, you should have:
python --versionDon’t worry if you’re new to async Python or web APIs - we’ll explain everything as we go!
# Install Zenith with all dependenciespip install zenithweb
# Create new project (or follow manual setup below)zen new taskflow # Complete project structurecd taskflow && zen dev # Running with hot reload!
# Includes:# - Pre-configured dependencies# - Best practice project structure# - Database models ready to use# - Authentication system# - API documentation# - Testing framework# - Security middlewareFor this tutorial, we’ll set up the project manually to understand each component:
uv is a fast Python package manager:
# Install uv if you don't have itcurl -LsSf https://astral.sh/uv/install.sh | sh
# Create our project with Zenithuv init taskflow-apicd taskflow-api
# Add Zenithuv add zenithweb
# Add development dependencies we'll needuv add --dev pytest pytest-asyncio black ruffTraditional approach with Python’s built-in tools:
# Create project directorymkdir taskflow-apicd taskflow-api
# Create virtual environmentpython -m venv venv
# Activate it (Linux/Mac)source venv/bin/activate# Or on Windows# venv\Scripts\activate
# Install Zenithpip install zenithweb
# Install dev dependenciespip install pytest pytest-asyncio black ruff
# Save dependenciespip freeze > requirements.txtLet’s organize our project properly from the start. Good structure makes everything easier as the project grows:
# Create our project structuremkdir -p app/{models,services,routes,utils}mkdir -p tests/{unit,integration}touch app/__init__.py app/{models,services,routes,utils}/__init__.pyYour project should now look like this:
taskflow-api/├── app/│ ├── __init__.py│ ├── main.py # Application entry point (we'll create this)│ ├── config.py # Configuration settings│ ├── models/ # Database models│ │ └── __init__.py│ ├── services/ # Business logic│ │ └── __init__.py│ ├── routes/ # API endpoints│ │ └── __init__.py│ └── utils/ # Helper functions│ └── __init__.py├── tests/ # Test files│ ├── unit/│ └── integration/├── .env # Environment variables (we'll create this)├── .gitignore # Git ignore file└── requirements.txt # Dependencies (if using pip)Now let’s create our application. Create app/main.py:
"""TaskFlow API - Task management system built with Zenith.
This is our main application file that creates and configuresthe Zenith app instance."""
from zenith import Zenithfrom datetime import datetime
# Create the Zenith application# In development, Zenith auto-configures based on environmentapp = Zenith( title="TaskFlow API", description="Task management system API", version="0.0.1")
# Our first endpoint - API root information@app.get("/")async def root(): """ API root endpoint.
Returns basic information about the API including available endpoints and current status. """ return { "name": "TaskFlow API", "version": "0.0.1", "status": "online", "timestamp": datetime.utcnow().isoformat(), "endpoints": { "api": "/", "health": "/health", "docs": "/docs" # Will be auto-generated! } }
# Health check endpoint - important for production@app.get("/health")async def health_check(): """ Health check endpoint for monitoring.
Used by: - Load balancers to check if the server is alive - Monitoring systems (Datadog, New Relic, etc.) - Container orchestrators (Kubernetes, ECS) """ return { "status": "healthy", "timestamp": datetime.utcnow().isoformat(), "service": "taskflow-api", "checks": { "api": "operational", # We'll add database check here later } }
# Development runnerif __name__ == "__main__": # This allows us to run the app directly with: python app/main.py import uvicorn uvicorn.run( "app.main:app", host="0.0.0.0", port=8000, reload=True # Auto-reload on code changes )Let’s break down what’s happening in our code:
app = Zenith( title="TaskFlow API", description="Task management system API", version="0.0.1")This creates our application instance. Zenith automatically:
async def root(): return {...}Notice the async keyword? This makes our function asynchronous, allowing it to:
@app.get("/")This decorator tells Zenith:
/)Let’s start our API! You have three ways to run it:
python app/main.pyuvicorn app.main:app --reloadzen serve --reloadYou should see output like:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)INFO: Started reloader process [28720] using WatchFilesINFO: Started server process [28722]INFO: Waiting for application startup.INFO: Application startup complete.Open a new terminal and let’s test our endpoints:
# Test the root endpointcurl http://localhost:8000/
# You should see:{ "name": "TaskFlow API", "version": "0.0.1", "status": "online", "timestamp": "2024-09-20T10:30:00", "endpoints": { "api": "/", "health": "/health", "docs": "/docs" }}
# Test the health endpointcurl http://localhost:8000/health# Install httpx first: pip install httpximport httpx
# Test our APIresponse = httpx.get("http://localhost:8000/")print(response.json())Open your browser and visit:
http://localhost:8000/ - See the JSON responsehttp://localhost:8000/docs - Interactive API documentation (auto-generated!)http://localhost:8000/redoc - Alternative documentation styleReal applications need configuration. Let’s create app/config.py:
"""Application configuration.
Loads settings from environment variables with sensible defaultsfor development."""
import osfrom typing import Optional
class Settings: """Application settings loaded from environment."""
# Application APP_NAME: str = os.getenv("APP_NAME", "TaskFlow API") APP_VERSION: str = os.getenv("APP_VERSION", "0.0.1") DEBUG: bool = os.getenv("DEBUG", "true").lower() == "true"
# Server HOST: str = os.getenv("HOST", "0.0.0.0") PORT: int = int(os.getenv("PORT", "8000"))
# Database (we'll use this in Part 2) DATABASE_URL: str = os.getenv( "DATABASE_URL", "sqlite:///./taskflow.db" # SQLite for development )
# Security (we'll use this in Part 4) # Production SECRET_KEY must have sufficient entropy (≥16 unique chars) # Generate with: zen keygen SECRET_KEY: str = os.getenv( "SECRET_KEY", "development-secret-change-in-production" )
# Email (we'll use this in Part 6) SMTP_HOST: Optional[str] = os.getenv("SMTP_HOST") SMTP_PORT: int = int(os.getenv("SMTP_PORT", "587")) SMTP_USER: Optional[str] = os.getenv("SMTP_USER") SMTP_PASSWORD: Optional[str] = os.getenv("SMTP_PASSWORD")
@property def is_production(self) -> bool: """Check if running in production.""" return not self.DEBUG
@property def is_development(self) -> bool: """Check if running in development.""" return self.DEBUG
# Create singleton instancesettings = Settings()Now create a .env file for local development:
APP_NAME=TaskFlow APIAPP_VERSION=0.0.1DEBUG=true
# DatabaseDATABASE_URL=sqlite:///./taskflow.db
# Security - CHANGE IN PRODUCTION!# Generate secure key with: zen keygen# Requirements: ≥32 chars, ≥16 unique chars, no char >25% frequencySECRET_KEY=dev-secret-key-change-this-in-production
# Email (optional for now)# SMTP_HOST=smtp.gmail.com# SMTP_PORT=587# SMTP_USER=your-email@gmail.com# SMTP_PASSWORD=your-app-passwordAnd update app/main.py to use our configuration:
from app.config import settings
app = Zenith( title=settings.APP_NAME, version=settings.APP_VERSION, debug=settings.DEBUG)Let’s add a simple feature - a placeholder for tasks. Update app/main.py:
# Temporary in-memory storage (we'll add a database in Part 2)tasks = []task_counter = 0
@app.get("/tasks")async def list_tasks( completed: bool = None, # Optional filter limit: int = 10 # Pagination limit): """ List all tasks with optional filtering.
Query Parameters: - completed: Filter by completion status - limit: Maximum number of tasks to return """ filtered_tasks = tasks
# Apply completed filter if provided if completed is not None: filtered_tasks = [ task for task in tasks if task["completed"] == completed ]
# Apply limit filtered_tasks = filtered_tasks[:limit]
return { "tasks": filtered_tasks, "total": len(filtered_tasks) }
@app.post("/tasks")async def create_task(task_data: dict): """ Create a new task.
For now, accepts any JSON. We'll add validation in Part 2. """ global task_counter task_counter += 1
# Create task with defaults task = { "id": task_counter, "title": task_data.get("title", "Untitled Task"), "description": task_data.get("description", ""), "completed": False, "created_at": datetime.utcnow().isoformat() }
tasks.append(task)
return { "message": "Task created successfully", "task": task }Test the new endpoints:
# Create a taskcurl -X POST http://localhost:8000/tasks \ -H "Content-Type: application/json" \ -d '{"title": "Learn Zenith", "description": "Complete the tutorial"}'
# List taskscurl http://localhost:8000/tasks
# Filter completed taskscurl http://localhost:8000/tasks?completed=falseSolution: Another process is using the port. Either:
lsof -i :8000 then kill <PID>uvicorn app.main:app --port 8001Solution: Make sure you’re in the virtual environment:
# Check if activated (you should see (venv) in prompt)which python
# If not, activate itsource venv/bin/activate # Linux/Mac# orvenv\Scripts\activate # WindowsSolution: Make sure you’re running with --reload:
uvicorn app.main:app --reloadIn this first part, you’ve:
In Part 2: Data Models, we’ll:
Continue to Part 2
Ready to add a database? Continue to Data Models →
Questions? Check our FAQ or ask in GitHub Discussions.