Securing CRUD Operations for Primary Resources

A Quick Recap of Our Architecture

As we add a new resource, let's remember the architectural pattern we're following. Each new feature, like our "Post" resource, will have its components distributed across the presentation, business, and data layers.

1. Setting Up the "Post" Resource

Before we can create endpoints, we need to define what a "Post" is in our system. This involves creating a database model and Pydantic schemas.

a. The Database Model

First, create a new file app/models/post.py to define the SQLAlchemy model for our posts. A post will have content and, crucially, a relationship to the user who created it.

app/models/post.py

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

from app.db.base_class import Base

class Post(Base):
    id = Column(Integer, primary_key=True, index=True)
    content = Column(String, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    



    # Foreign key to the User model
    owner_id = Column(Integer, ForeignKey("user.id"))
    



    # Relationship to load the owner
    owner = relationship("User", back_populates="posts")

We also need to update our User model to complete the relationship.

app/models/user.py (Update)

# ... other imports
from sqlalchemy.orm import relationship

class User(Base):



    # ... other columns
    



    # Add this line to define the other side of the relationship
    posts = relationship("Post", back_populates="owner")

Now, generate and apply the database migration with Alembic:

uv run alembic revision --autogenerate -m "Add post model"
uv run alembic upgrade head

b. Pydantic Schemas

Next, define the data shapes for our API. Create app/schemas/post.py. We need different schemas for creating, updating, and viewing posts.

app/schemas/post.py

from pydantic import BaseModel
from datetime import datetime




# Shared properties
class PostBase(BaseModel):
    content: str




# Schema for creating a post
class PostCreate(PostBase):
    pass




# Schema for updating a post
class PostUpdate(PostBase):
    pass




# Schema for reading/returning a post from the API
class PostPublic(PostBase):
    id: int
    owner_id: int
    created_at: datetime

    class Config:
        orm_mode = True # In Pydantic v2, this is from_attributes = True

Using separate schemas gives you fine-grained control over your API contract, a concept that parallels using different serializers for read and write operations in Django REST Framework.

2. Implementing the CRUD Endpoints

With our models and schemas ready, let's create the repository, service, and router for our Post resource.

First, create the placeholder files:

• app/repositories/post_repository.py
• app/services/post_service.py
• app/routers/post_router.py

Now, we'll implement the CRUD logic step-by-step.

Create (C): Creating a New Post

A user must be logged in to create a post. We'll use our get_current_user dependency to enforce this.

app/repositories/post_repository.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

from app.models.post import Post
from app.schemas.post import PostCreate

class PostRepository:
    def __init__(self, db_session: AsyncSession):
        self.db_session = db_session

    async def create_post(self, post_data: PostCreate, owner_id: int) -> Post:
        new_post = Post(content=post_data.content, owner_id=owner_id)
        self.db_session.add(new_post)
        await self.db_session.commit()
        await self.db_session.refresh(new_post)
        return new_post

app/services/post_service.py

from app.repositories.post_repository import PostRepository
from app.schemas.post import PostCreate
from app.models.user import User
from app.models.post import Post

class PostService:
    def __init__(self, post_repo: PostRepository):
        self.post_repo = post_repo

    async def create_new_post(self, post_data: PostCreate, current_user: User) -> Post:
        return await self.post_repo.create_post(post_data, owner_id=current_user.id)

Notice how the PostService receives the current_user object and uses its ID to link the post to its owner. The repository only handles the database interaction.

Finally, the router ties it all together.

app/routers/post_router.py

from fastapi import APIRouter, Depends, HTTPException
from typing import List

from app.dependencies import get_post_service, get_current_user
from app.services.post_service import PostService
from app.schemas.post import PostCreate, PostPublic
from app.models.user import User

router = APIRouter(prefix="/posts", tags=["Posts"])

@router.post("/", response_model=PostPublic, status_code=201)
async def create_post(
    post_in: PostCreate,
    post_service: PostService = Depends(get_post_service),
    current_user: User = Depends(get_current_user),
):
    """
    Create a new post. Must be authenticated.
    """
    new_post = await post_service.create_new_post(post_in, current_user)
    return new_post

Don't forget to create the get_post_service dependency in app/dependencies.py just like you did for the other services!

By adding current_user: User = Depends(get_current_user), we have secured this endpoint. This is the FastAPI equivalent of adding permission_classes = [IsAuthenticated] to a DRF ViewSet. The dependency injection is explicit and type-hinted.

Read (R): Fetching Posts

Reading posts should be public. Anyone, authenticated or not, can view them. Therefore, we won't add the get_current_user dependency to these endpoints.

Let's add the methods to our repository and service first.

app/repositories/post_repository.py (additions)

# ...

    async def get_post_by_id(self, post_id: int) -> Post | None:
        return await self.db_session.get(Post, post_id)

    async def get_all_posts(self, skip: int = 0, limit: int = 100) -> List[Post]:
        result = await self.db_session.execute(
            select(Post).offset(skip).limit(limit)
        )
        return result.scalars().all()

app/services/post_service.py (additions)

# ...

    async def get_post(self, post_id: int) -> Post | None:
        return await self.post_repo.get_post_by_id(post_id)
        
    async def get_all_posts(self, skip: int, limit: int) -> List[Post]:
        return await self.post_repo.get_all_posts(skip, limit)

Now, add the endpoints to app/routers/post_router.py.

app/routers/post_router.py (additions)

# ...

@router.get("/", response_model=List[PostPublic])
async def read_posts(
    skip: int = 0,
    limit: int = 100,
    post_service: PostService = Depends(get_post_service),
):
    """
    Retrieve all posts. Public endpoint.
    """
    posts = await post_service.get_all_posts(skip=skip, limit=limit)
    return posts

@router.get("/{post_id}", response_model=PostPublic)
async def read_post(
    post_id: int,
    post_service: PostService = Depends(get_post_service),
):
    """
    Retrieve a single post by its ID. Public endpoint.
    """
    post = await post_service.get_post(post_id=post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

Update (U) & Delete (D): Enforcing Ownership

Updating and deleting a post are sensitive operations. Not only must the user be authenticated, but they must also be the owner of the post. This is an authorization check.

In DRF, you would typically create a custom permission class like IsOwnerOrReadOnly. In FastAPI, we can implement this logic directly in our service layer.

Let's see how this works for the delete operation.

app/repositories/post_repository.py (addition)

# ...
    async def delete_post(self, post_id: int) -> None:
        post = await self.db_session.get(Post, post_id)
        if post:
            await self.db_session.delete(post)
            await self.db_session.commit()

app/services/post_service.py (addition)

# ...

    async def delete_post(self, post_id: int, current_user: User):



        # 1. Fetch the post
        post_to_delete = await self.post_repo.get_post_by_id(post_id)

        if not post_to_delete:
            raise HTTPException(status_code=404, detail="Post not found")




        # 2. Perform the authorization check
        if post_to_delete.owner_id != current_user.id:
            raise HTTPException(
                status_code=403, detail="Not authorized to delete this post"
            )




        # 3. If authorized, proceed with deletion
        await self.post_repo.delete_post(post_id)

The authorization logic is explicit: fetch the resource, then check if the resource's owner ID matches the current user's ID.

Now, let's wire this up in the router.

app/routers/post_router.py (addition)

from fastapi import status



# ...

@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(
    post_id: int,
    post_service: PostService = Depends(get_post_service),
    current_user: User = Depends(get_current_user),
):
    """
    Delete a post. Must be the owner.
    """
    await post_service.delete_post(post_id, current_user)
    return None # Return None for 204 status code

This pattern is very powerful. The dependency injection handles authentication (get_current_user), and the service layer handles the business-specific authorization logic.

You can now implement the update functionality using the exact same pattern: add update_post methods to the repository and service, fetch the post, check for ownership, and if the check passes, apply the updates and commit. I encourage you to implement this on your own as an exercise.

Conclusion

Congratulations! You have successfully implemented a full set of secured CRUD endpoints for a primary resource, bringing our capstone project to life. You've seen how FastAPI's features, combined with a clean architecture, allow for explicit and powerful control over both authentication and authorization.

Key Takeaways:

• CRUD Implementation: We built endpoints for Creating, Reading, Updating, and Deleting posts, following the service-repository pattern.
• Authentication vs. Authorization: We used the Depends(get_current_user) dependency to handle authentication (is the user logged in?). We implemented custom logic in the service layer to handle authorization (is this user allowed to perform this action on this specific resource?).
• Analogies to Django/DRF:
  • Securing an endpoint with Depends(get_current_user) is like using permission_classes = [IsAuthenticated].
  • The manual ownership check (if post.owner_id != current_user.id) is equivalent to creating a custom IsOwnerOrReadOnly permission class in DRF.
• Separation of Concerns: The router handles HTTP requests/responses, the service contains business and authorization logic, and the repository manages database interactions.

Preview of the Next Lesson:

Now that we have a working API for creating posts, what if we wanted to perform an action after the post is created, without making the user wait? For example, analyzing the post content for sentiment or notifying other users. In the next lesson, we will explore FastAPI's BackgroundTasks to handle these "fire-and-forget" operations, comparing this approach to more robust solutions like Celery.