Writing Clean Code with FastAPI Dependency Injection

#python #fastapi #programming #tutorial

Before anyone else read the code that I wrote, “clean code” wasn’t really a thing for me. I had no issues reading my code because, well, I wrote it.

That was also pretty true for me in college as well. Classes rarely worried about the readability of code - instead stressing correctness and performance.

At my first job, however, they pretty much immediately showed that the opposite was important. Subtle, clever code is way worse than obvious, slow code because you now have a whole team of people who will need to read it, understand it, update it, etc. There are exceptions, of course, but for most of the code I’ve written professionally, it’s way better to be obvious than clever.

In this post, I wanted to show the process of taking a difficult to read route and improving it. We will not change the functionality at all, but instead just make it easier for the next person to understand.

To make it a bit more realistic, we’re going to use a FastAPI route as an example, and we’re also going to use FastAPI’s dependency injection, which can really help with readability (and testability, but more on that later).

Let’s start with a FastAPI route:

class UsernameInput(BaseModel):
    name: str

@app.post("/games/{game_id}/join")
def join_game(game_id: str, username: UsernameInput):
    if not username.name:
        raise HTTPException(status_code=400, detail="Username is required")
    elif len(username.name) > 20:
        raise HTTPException(status_code=400, detail="Username must be less than 20 characters")
    elif not username.name.isalnum():
        raise HTTPException(status_code=400, detail="Username must be alphanumeric")

    with Session(engine) as session:
        # find the game
        select_query = select(Game).where(Game.id == game_id)
        game = session.exec(select_query).first()
        if not game:
            raise HTTPException(status_code=404, detail="Game not found")

        # check if the game is open, and add the player
        if game.status != "open":
            raise HTTPException(status_code=400, detail="Game is not open")
        game.players.append(Player(name=username.name))
        session.commit()
        session.refresh(game)

    return {"game_id": game.id, "start_time": game.start_time}

This is using SQLModel to talk to the DB (if you are interested in a deeper dive on that, see this post).

This route isn’t terrible, but it does take a second and some reading to figure out all the details. If you had a file with a few of these routes, it’ll get long quickly.

The first thing I like to do when refactoring code like this is to try and just use code actions built into my editor. Most commonly, this is extract method (pictured below is JetBrains but VSCode also has the same functionality - and you can add it to Neovim):

What’s nice about this is I can be confident the code isn’t changing, and the end result looks like this:

# TODO: all the methods we pull out could go into their own files
def validate_username(username: UsernameInput):
    if not username.name:
        raise HTTPException(status_code=400, detail="Username is required")
    elif len(username.name) > 20:
        raise HTTPException(status_code=400, detail="Username must be less than 20 characters")
    elif not username.name.isalnum():
        raise HTTPException(status_code=400, detail="Username must be alphanumeric")

@app.post("/games/{game_id}/join")
def join_game(game_id: str, username: UsernameInput):
    validate_username(username)

    with Session(engine) as session:
        # find the game
        select_query = select(Game).where(Game.id == game_id)
        game = session.exec(select_query).first()
        if not game:
            raise HTTPException(status_code=404, detail="Game not found")

        # check if the game is open, and add the player
        if game.status != "open":
            raise HTTPException(status_code=400, detail="Game is not open")
        game.players.append(Player(name=username.name))
        session.commit()
        session.refresh(game)

    return {"game_id": game.id, "start_time": game.start_time}

Next, we can take our two comments and turn them into functions with Extract Method too:

def find_open_game(session: Session, game_id: str):
    select_query = select(Game).where(Game.id == game_id)
    game = session.exec(select_query).first()
    if not game:
        raise HTTPException(status_code=404, detail="Game not found")
    if game.status != "open":
        raise HTTPException(status_code=400, detail="Game is not open")
    return game

def add_player_to_game(session: Session, game: Game, username: UsernameInput):
    game.players.append(Player(name=username.name))
    session.commit()
    session.refresh(game)    
    return game

And now, if we look back at our route, it’s a lot easier to parse quickly:

@app.post("/games/{game_id}/join")
def join_game(game_id: str, username: UsernameInput):
    validate_username(username)

    with Session(engine) as session:
        game = find_open_game(session, game_id)
        game = add_player_to_game(session, game, username)

    return {"game_id": game.id, "start_time": game.start_time}

We haven’t changed what the code does - all the complexity still exists. The difference is that the reader isn’t greeted with it, they can opt-into it by looking at the related functions.

But we can actually take this a few steps further with FastAPI’s dependency injection. We’ll start by moving our session out of route itself so every one of our routes doesn’t need to be indented:

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

@app.post("/games/{game_id}/join")
def join_game(*, 
              session: Session = Depends(get_session), 
              game_id: str, 
              username: UsernameInput):
    validate_username(username)
    game = find_open_game(session, game_id)
    game = add_player_to_game(session, game, username)
    return {"game_id": game.id, "start_time": game.start_time}

Next, let’s look at the username. We’re taking in a UsernameInput and validating it in the route. As our company grows and more developers are adding routes, it’s possible that someone will forget to add that validate_username call.

What if we instead move that logic into a reusable dependency:

def get_validated_username(username: UsernameInput):
    validate_username(username)
    return username.name

@app.post("/games/{game_id}/join")
def join_game(*, 
              session: Session = Depends(get_session), 
              game_id: str, 
              username: str = Depends(get_validated_username)):
    game = find_open_game(session, game_id)
    game = add_player_to_game(session, game, username)
    return {"game_id": game.id, "start_time": game.start_time}

If we want to be extra fancy (and who doesn’t?), we could have a separate type for the validated username, and all our functions would use that instead of strings or UsernameInputs.

The game_id can get similar treatment. One caveat here is that we’ll probably have other use cases for finding a game, not just an open game. Instead of re-using find_open_game we’ll make a new version that doesn’t check if the game is open:

# If session is depended on multiple times in the same request, it'll only be called once
def fetch_game(*, session: Session = Depends(get_session), game_id: str):
    select_query = select(Game).where(Game.id == game_id)
    game = session.exec(select_query).first()
    if not game:
        raise HTTPException(status_code=404, detail="Game not found")
    return game

And then we can use that directly:

@app.post("/games/{game_id}/join")
def join_game(*, 
              session: Session = Depends(get_session), 
              game: Game = Depends(fetch_game), 
              username: str = Depends(get_validated_username)):
    if game.status != "open":
        raise HTTPException(status_code=400, detail="Game is not open")
    game = add_player_to_game(session, game, username)
    return {"game_id": game.id, "start_time": game.start_time}

The body of our function is completely focused on just the core, unique functionality of join_game

All those common edge cases that we were handling in our code are now hidden. If someone submits an invalid username, it’s automatically rejected. If a game_id doesn’t correspond to a real game, a 404 is returned.

Maybe going a little overboard

Before we keep going, I should say this last route is a totally valid stopping point. Clean code is subjective and it’s 100% possible to overdo it.

However, if we were to keep going, here’s another option that removes the need for the session and game dependencies:

class GameManager:
    def __init__(self, session: Session, game: Game):
        self.session = session
        self.game = game

    def is_open(self):
        return self.game.status == "open"

    def add_player(self, username: str):
        self.game.players.append(Player(name=username))
        self.session.commit()
        self.game = self.session.refresh(self.game)

def fetch_game_manager_by_id(*, 
                             session: Session = Depends(get_session),
                             game: Game = Depends(fetch_game)):
    return GameManager(session, game)

and our final route would look like:

@app.post("/games/{game_id}/join")
def join_game(*,
              game: GameManager = Depends(fetch_game_manager_by_id),
              username: str = Depends(get_validated_username)):
    if !game.is_open():
        raise HTTPException(status_code=400, detail="Game is not open")
    game.add_player(username)
    return {"game_id": game.game.id, "start_time": game.game.start_time}

For me, this is approaching the point of “too much magic.” You might instead want the GameManager to not have it’s own game but instead get a game passed in to it - but both approaches are pretty reasonable.

A nice perk: Testability of our FastAPI routes

Initially, we talked about this as a means to make our code more readable. However, clean code and testable code often go hand in hand. By modeling our route with dependencies that get injected, we could override those dependencies with fake/mock implementations.

For one example, if we wanted to write a test called players_cant_join_closed_game, an easy way to do that is to create a mock GameManager where .is_open() always returns False. This allows us to test that the route is respecting the interfaces we created, and we don’t even need a DB to do it.

You likely still want to have integration and end-to-end tests that do use the database. However, even in that case, you may still want to mock other external resources (like email sending), and injecting those dependencies makes it easy to override.

When clean code goes wrong

I mentioned that it’s possible to overdo it, let’s see what that looks like:

@app.post("/games/{game_id}/join")
def join_game(*,
              game: Game = Depends(fetch_game_by_id),
              game_open_checker: GameOpenChecker = Depends(get_game_open_checker),
              game_membership_manager: GameMembershipManager = Depends(get_game_membership_manager),
              game_result_formatter: GameResultFormatter = Depends(get_game_result_formatter),
              player_creator: PlayerCreator = Depends(get_player_creator),
              username: str = Depends(get_validated_username)):
    if !game_open_checker.is_open(game):
        raise HTTPException(status_code=400, detail="Game is not open")
    player = player_creator.create(username)
    game = game_membership_manager.add_player(player)
    return game_result_formatter.format_join_game_response(game)

To me, this is a step back in readability. We really don’t need a dependency for everything, it can be beneficial to group them together like we did before.

Clean code is a balancing act - you’ll want to make sure you don’t turn your codebase into something like this.

Summary

One of the best ways to start writing more readable code is just to try and break your functions down into their logical pieces. Your IDE can help you with code actions like Extract Method or Change Signature. After that, you can start to move shared functionality into re-usable dependencies and make it significantly easier for future readers (and your future self!) to parse at a glance.