RealWorldApp - part 3

• September 28, 2025
• 961 words
• 5 min

Hi,

after RealWorldApp part 1 and part 2 comes part 3.

In part 2 we implemented our data model.

Let's check that it works!

Dummy data

Using Mistral AI we created dummy data to populate our database.

Test time

Database session

We'll need a database session dependency to manage our connections to the database. We define it in the app/db.py for the moment and put the definition of the functions responsible of dabase creation and deletion :

# app/db.py
from typing import Annotated

from fastapi import Depends
from sqlmodel import Session, SQLModel, create_engine

from app import models

sqlite_file_name = "database.db"
slqite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(slqite_url)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def drop_db():
    SQLModel.metadata.drop_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


DbSessionDep = Annotated[Session, Depends(get_session)]

First creation of the FastAPI app

(finaly)

Currently our project is structured like this:

.
├── app
│   ├── db.py
│   ├── __init__.py
│   ├── main.py
│   └── models.py
├── CODE_OF_CONDUCT.md
├── database.db
├── favicon.ico
├── LICENSE
├── logo.png
├── pyproject.toml
├── README.md
└── uv.lock

Let's change app/main.py to finally create a FastAPI app. We keep the previous ability to create the database but keep it away and a drop database could also be useful:

from fastapi import FastAPI


app = FastAPI()

Now let's add some route to read our database.

Get Users

from fastapi import FastAPI, HTTPException
from sqlmodel import select

from app.db import DbSessionDep
from app.models import User
# ...
app = FastAPI()


@app.get("/api/users")
async def read_users(db_session: DbSessionDep):
    users = db_session.exec(select(User)).all()
    return users


@app.get("/api/user/{username}")
async def read_user(username: str, db_session: DbSessionDep):
    user = db_session.exec(select(User).where(User.username == username)).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Get Articles

# ...
from sqlalchemy.orm import aliased

from app.models import Article, User
# ...
@app.get("/api/articles")
async def read_articles(
    db_session: DbSessionDep,
    tag: str | None = None,
    author: str | None = None,
    favorited: str | None = None,
    limit: int | None = 20,
    offset: int | None = 0,
):
    statement = select(Article).offset(offset).limit(limit)
    if tag:
        statement = statement.join(ArticlesHaveTags).join(Tag).where(Tag.name == tag)
    if author:
        user_author = aliased(User, name="u_author")
        statement = statement.join(user_author).where(user_author.username == author)
    if favorited:
        fan = aliased(User, name="u_fan")
        statement = (
            statement.join(
                UsersFavourArticles,
                Article.id == UsersFavourArticles.article_id,
            )
            .join(fan)
            .where(fan.username == favorited)
        )
    articles = db_session.exec(statement).all()
    return articles


@app.get("/api/article/{slug}")
async def read_article(slug: str, db_session: DbSessionDep):
    article = db_session.exec(select(Article).where(Article.slug == slug)).first()
    if not article:
        raise HTTPException(status_code=404, detail="article not found")
    return article.tags

Get Tags

# ...
from app.models import Article, User, Tag
# ...
@app.get("/api/tags")
async def read_tags(db_session: DbSessionDep):
    tags = db_session.exec(select(Tag)).all()
    return tags


@app.get("/api/tag/{tag_id}")
async def read_tag(tag_id: int, db_session: DbSessionDep):
    tag = db_session.exec(select(Tag).where(Tag.id == tag_id)).first()
    if not tag:
        raise HTTPException(status_code=404, detail="tag not found")
    return tag

Get Comments from an article

# ...
from app.models import Article, User, Tag,
# ...
@app.get("/api/articles/{slug}/comments")
async def read_comments_from_article(slug: str, db_session: DbSessionDep):
    article = db_session.exec(select(Article).where(Article.slug == slug)).first()
    if not article:
        raise HTTPException(status_code=404, detail="article not found")
    comments = article.commenting_users
    return comments

Next

Great! We can read our data. In the next session we'll implement authentication.