Blog Tutorial
Build a complete blog application with Ryx — models, relationships, queries, and more.
Step 1: Define Models
from ryx import (
Model, CharField, TextField, DateTimeField,
ForeignKey, BooleanField, SlugField,
Index,
)
class Author(Model):
name = CharField(max_length=100)
email = CharField(max_length=254, unique=True)
bio = TextField(null=True, blank=True)
class Meta:
ordering = ["name"]
class Category(Model):
name = CharField(max_length=50, unique=True)
description = TextField(null=True, blank=True)
slug = SlugField(unique=True)
class Post(Model):
title = CharField(max_length=200)
slug = SlugField(max_length=210, unique=True)
body = TextField()
active = BooleanField(default=False)
author = ForeignKey(Author, on_delete="CASCADE", related_name="posts")
category = ForeignKey(Category, on_delete="SET_NULL", null=True, related_name="posts")
created_at = DateTimeField(auto_now_add=True)
updated_at = DateTimeField(auto_now=True)
class Meta:
ordering = ["-created_at"]
indexes = [Index(fields=["slug"], name="idx_post_slug")]
async def clean(self):
if self.active and not self.body:
raise ValidationError({"body": ["Active posts must have content"]})
class Comment(Model):
post = ForeignKey(Post, on_delete="CASCADE", related_name="comments")
author = CharField(max_length=100)
body = TextField()
created_at = DateTimeField(auto_now_add=True)
class Meta:
ordering = ["created_at"]
Step 2: Setup & Migrate
import asyncio
import ryx
from ryx.migrations import MigrationRunner
async def setup():
await ryx.setup("sqlite:///blog.db")
await MigrationRunner([Author, Category, Post, Comment]).migrate()
asyncio.run(setup())
Step 3: Create Content
# Create authors
alice = await Author.objects.create(name="Alice", email="alice@example.com")
bob = await Author.objects.create(name="Bob", email="bob@example.com")
# Create categories
tech = await Category.objects.create(name="Technology", slug="tech")
life = await Category.objects.create(name="Lifestyle", slug="life")
# Create posts
post1 = await Post.objects.create(
title="Getting Started with Rust",
slug="getting-started-rust",
body="Rust is a systems programming language...",
active=True,
author=alice,
category=tech,
)
post2 = await Post.objects.create(
title="My Morning Routine",
slug="morning-routine",
body="I wake up at 5am every day...",
active=True,
author=bob,
category=life,
)
# Create comments
await Comment.objects.create(post=post1, author="Reader1", body="Great article!")
await Comment.objects.create(post=post1, author="Reader2", body="Very helpful!")
await Comment.objects.create(post=post2, author="Reader3", body="Inspiring!")
Step 4: Query
# All active posts
posts = await Post.objects.filter(active=True)
# Posts by a specific author
alice_posts = await Post.objects.filter(author_id=alice.pk)
# Posts in a category with comment count
from ryx import Count
posts_with_comments = await (
Post.objects
.filter(active=True)
.annotate(comment_count=Count("comments.id"))
.order_by("-comment_count")
)
# Search posts
posts = await Post.objects.filter(
Q(title__contains="Rust") | Q(body__contains="Rust"),
active=True,
)
# Recent posts with author info
from ryx.relations import apply_select_related
posts = await Post.objects.filter(active=True).order_by("-created_at").limit(5)
posts = await apply_select_related(posts, fields=["author", "category"])
for post in posts:
print(f"{post.title} by {post.author.name} in {post.category.name}")
Step 5: Stats Dashboard
from ryx import Count, Sum
# Blog-wide stats
stats = await Post.objects.aggregate(
total_posts=Count("id"),
active_posts=Count("id", filter=Q(active=True)),
total_comments=Count("comments.id"),
)
# Per-author stats
author_stats = await (
Post.objects
.values("author_id", "author__name")
.annotate(posts=Count("id"))
.order_by("-posts")
)
# Per-category stats
category_stats = await (
Post.objects
.filter(active=True)
.values("category_id", "category__name")
.annotate(posts=Count("id"))
)
Step 6: Signals & Hooks
from ryx import receiver, post_save, pre_delete
@receiver(post_save, sender=Comment)
async def on_comment_saved(sender, instance, created, **kwargs):
if created:
print(f"New comment on post #{instance.post_id}")
@receiver(pre_delete, sender=Author)
async def on_author_deleting(sender, instance, **kwargs):
post_count = await Post.objects.filter(author_id=instance.pk).count()
if post_count > 0:
print(f"Warning: {instance.name} has {post_count} posts that will be deleted")
Complete Script
See examples/ in the repository for runnable versions of each step.
Next Steps
→ Testing — Test strategies with Ryx