Social network

Murmur: posts, likes, follows, profiles and a home timeline.

social-network/
├── app.py            # composition root
├── settings.py       # NETWORK_NAME, MAX_POST_LENGTH, DATABASE_URL
├── models/           # database.py, member.py (Member, Follow), post.py (Post, Like)
├── services/         # social.py (SocialService), seed.py
└── views/
    ├── components/   # shell, composer, post_card, who_to_follow
    ├── pages/        # home, explore, profile, account
    └── styles/

1. Association tables

class Follow(Base):
    __tablename__ = "follows"

    follower_id: Mapped[int] = mapped_column(ForeignKey("members.id", ondelete="CASCADE"), primary_key=True)
    followee_id: Mapped[int] = mapped_column(ForeignKey("members.id", ondelete="CASCADE"), primary_key=True)

class Like(Base):
    __tablename__ = "likes"

    member_id: Mapped[int] = mapped_column(ForeignKey("members.id", ondelete="CASCADE"), primary_key=True)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True, index=True)

Composite primary keys make “follow twice” and “like twice” impossible at the database level.

2. Timelines in a fixed number of queries

    @remote(method="GET", auth="required")
    async def timeline(self, limit: int = 50) -> list[PostView]:
        """Your posts and the posts of the people you follow, newest first."""
        async with self.db.session() as db:
            me = await self._me(db)
            followees = select(Follow.followee_id).where(Follow.follower_id == me.id)
            query = _posts().where(or_(Post.author_id == me.id, Post.author_id.in_(followees)))
            return await self._views(db, list(await db.scalars(query.limit(min(limit, 100)))), me.id)

    async def _views(self, db: AsyncSession, posts: list[Post], me_id: int | None) -> list[PostView]:
        ids = [p.id for p in posts]
        like_counts = select(Like.post_id, func.count()).where(Like.post_id.in_(ids)).group_by(Like.post_id)
        counts = dict((await db.execute(like_counts)).all())
        liked = set()
        if me_id is not None:
            liked = set(await db.scalars(select(Like.post_id).where(Like.member_id == me_id, Like.post_id.in_(ids))))
        return [
            PostView(
                id=p.id,
                body=p.body,
                created_at=p.created_at,
                author_handle=p.author.handle,
                author_name=p.author.name,
                likes=counts.get(p.id, 0),
                liked=p.id in liked,
            )
            for p in posts
        ]

Authors come in one selectinload query, like counts in one GROUP BY, and your likes in one IN query, whether the timeline has 5 posts or 500. The test suite counts the SQL statements to prove it.

3. Local updates

class PostCard(Component):
    """One post. Liking updates this card only: its state starts from the props and then
    follows the server's answer, without reloading the whole timeline."""

    post: PostView
    can_like: bool = False
    likes = State(0)
    liked = State(False)

    def mount(self):
        self.likes, self.liked = self.post.likes, self.post.liked

    async def toggle_like(self):
        updated = await self.use(SocialService).toggle_like(self.post.id)
        self.likes, self.liked = updated.likes, updated.liked

    def render(self):
        p = self.post
        hearts = f"♥ {self.likes}"
        count = f"{self.likes} like{'s' if self.likes != 1 else ''}"
        like = (
            Button(
                hearts,
                size="sm",
                variant="secondary" if self.liked else "ghost",
                aria_pressed="true" if self.liked else "false",
                aria_label=f"{'Unlike' if self.liked else 'Like'} post by {p.author_name} ({count})",
                on_click=self.toggle_like,
            )
            if self.can_like
            else Text(hearts, muted=True, size="sm", aria_label=count)
        )
        return Card(
            Row(
                Avatar(p.author_name, size=40),
                Column(
                    Row(
                        Link(p.author_name, href=f"/u/{p.author_handle}", style=Style(font_weight=700)),
                        Text(f"@{p.author_handle} · {ago(p.created_at)}", muted=True, size="sm", tag="span"),
                        gap="xs",
                        wrap=True,
                    ),
                    Text(p.body, style=Style(white_space="pre-wrap", raw={"overflow-wrap": "anywhere"})),
                    Row(Spacer(), like),
                    gap="xs",
                    style=Style(grow=1, min_width=0),
                ),
                align="start",
                gap="sm",
            ),
            style=PANEL,
        )

Liking re-renders only that card: its state starts from the props in mount() and then follows the server’s answer.

4. A live character counter

class Composer(Component):
    """Write a post. The counter follows your typing (debounced) and the button only enables
    when the post is valid; the server validates again on submit, because it never trusts the browser."""

    posted: Callable[[], Awaitable[None]]
    draft = State("")
    version = State(0)

    def set_draft(self, value: str):
        self.draft = value

    async def publish(self, form: ComposeForm):
        await self.use(SocialService).publish(form.body)
        self.draft = ""
        self.version += 1  # a fresh form: the text area is cleared
        await self.posted()

    def render(self):
        left = MAX_POST_LENGTH - len(self.draft)
        valid = bool(self.draft.strip()) and left >= 0
        return Card(
            Form(
                Textarea(
                    "What's happening?",
                    name="body",
                    rows=3,
                    hide_label=True,
                    placeholder="What's happening?",
                    on_input=self.set_draft,
                ),
                Row(
                    Spacer(),
                    Text(
                        str(left),
                        size="sm",
                        color="danger" if left < 0 else "muted",
                        aria_label=f"{left} characters left",
                    ),
                    Button("Post", type="submit", disabled=not valid),
                ),
                on_submit=self.publish,
                key=f"compose-{self.version}",
                style=Style(gap="sm"),
            ),
            style=PANEL,
        )

The counter follows your typing and the button enables only for a valid post. The server still validates on submit, because it never trusts the browser.

5. Callbacks between components

@router.page("/", auth="required")
class Home(Page):
    """Your timeline, a composer, and suggestions beside it (below it on narrow screens)."""

    title = f"Home · {NETWORK_NAME}"
    posts = State(factory=list)

    async def mount(self):
        await self.reload()

    async def reload(self):
        self.posts = await self.use(SocialService).timeline()

    def render(self):
        return Shell(
            Row(
                Column(
                    Heading("Home", level=1),
                    Composer(posted=self.reload),
                    PostList(self.posts, can_like=True, empty="Follow people to fill your timeline."),
                    gap="md",
                    style=Style(grow=1, min_width=0),
                ),
                Aside(
                    WhoToFollow(followed=self.reload),
                    style=NARROW_SIDEBAR,
                ),
                align="start",
                gap="lg",
                collapse="narrow",
            )
        )

Composer(posted=self.reload) and WhoToFollow(followed=self.reload) tell the page to reload the timeline.