Building Video Game Recommender Systems with FastAPI, PostgreSQL, and Render: Part 1

Introduction

enable an application to generate intelligent suggestions for a user, effectively sorting relevant content out from the rest. In this article, we build and deploy a dynamic video game recommender system leveraging PostgreSQL, FastAPI, and Render to recommend new games for a user based on those they’ve interacted with. The intent is to provide a clear example of how a standalone recommender system can be constructed, which could then be tied into a front-end system or other application.

For this project, we use video game data accessible from Steams API but this could easily be replaced by whatever product data you’re interested in, the key steps will be the same. We’ll cover how to store this data in a Database, vectorize the game Tags, generate similarity scores based on the games a user has interacted with, and return a series of relevant recommendations. At the end of this article, we’ll have this recommender system deployed as a Web Application with FastAPI such that whenever a user interacts with a new Game, we can dynamically generate and store a new set of recommendations for that user.

The following tools will be used:

• PostgreSQL
• FastAPI
• Docker
• Render

Those just interested in the GitHub repository can find it here.

Table of Contents

Due to the length of this project, it’s divided into two articles. The first portion covers the setup and theory behind this project (steps 1–5 shown below), and the second part covers deploying it. If you’re looking for the second part it’s located here.

Part 1

1. Dataset Overview
2. Overall System Architecture
3. Database Setup
4. FastAPI Setup
   – Models
   – Routes
5. Building Similarity Pipeline

1. Deploying a PostgreSQL database on Render
2. Deploying a FastAPI app as a Render Web Application
   – Dockerizing our application
   – Pushing Docker Image to DockerHub
   – Pulling from DockerHub to Render

Dataset Overview

The dataset for this project contains data for the top ~2000 games from the steamworks API. This data is free and licensed for personal and commercial use, subject to the terms of service, there is a 200 requests/5 minute rate limit that resulted in us working with only a subset of the data. The terms of service can be found here.

An overview of the games dataset is shown below. Most of the fields are relatively self-descriptive; the key thing to note is that the unique product identifier is appid. In addition to this dataset, we also have several additional tables that we’ll detail below; the most important one for our recommender system is a game tags table, which contains the appid values mapped to each tag associated with the game (strategy, RPG, card game, etc.). These were drawn from the categories field shown in the Data Overview and then pivoted to create the game_tags table so that there’s a unique row for each appip:category combination.

For a more detailed overview of the structure of our project, see the diagram below.

Now we’ll provide a quick overview of the architecture of this project and then dive into how to populate our database.

Architecture

For our recommender system system, we’ll use a PostgreSQL database with a FastAPI data access + processing layer that will allow us to add or remove games from a user’s game list. Users making changes to their game library, via a FastAPI POST request, will also kick off a recommendation pipeline leveraging FastAPI’s Background Tasks function that will query their liked games from the database, calculate a similarity score with non-liked games, and update a user_recommendation table with their new top-N recommended games. Finally, both the PostgreSQL database and FastAPI service will be deployed on Render so they can be accessed beyond our local environment. For this deployment step, any cloud service could have been used, but we chose Render in this case for its simplicity.

To recap, our overall workflow from the user’s perspective will look like this:

1. The user adds a game to their library by making a POST request from FastAPI to our database.
   • If we wanted to attach our recommender system to a front-end application, we could easily tie this Post API into a user interface.
2. This post request kicks off a FastAPI background task that runs our recommender pipeline.
3. The recommender pipeline queries our database for the user’s game list and the global games list.
4. A similarity score is then calculated between the user’s games and all games using our game tags.
5. Finally, our recommender pipeline makes a post request to the database to update the recommended games table for that user.

Setting Up the Database

Before we build our recommender system, the first step is to set up our database. Our basic database diagram is shown in Figure 5. We previously discussed our game table above; this is the base dataset that the rest of our data is generally derived from. A full list of our tables is here:

• Game Table: Contains base game data for each unique game in our database
• User Table: A Dummy user table containing example information populated for example.
• User_Game Table: Contains the mappings between all games that a user has ‘liked’; this table is one of the base tables used to generate recommendations by capturing what games a user is interested in.
• Game_Tags Table: This contains an appid:game_tag mapping, where game tag could be something like ‘strategy’, ‘rpg’, ‘comedy’, a descriptive tag that captures part of the essence of a game. There are multiple tags mapped to each appid.
• User_Recommendation Table: This is our target table that will be updated by our pipeline. Every time a user interacts with a new game, our recommendation pipeline will run and generate a new series of recommendations for that user that will be stored here.

To set up these tables, we can simply run our src/load_database.py file. This file creates and populates our tables in a couple of steps that are outlined below. Note, right now we’re going to focus on understanding how to write this data to a generic database, so all you have to know now is that the External_Database_Url below is the URL to whatever database you want to use. In the second half of this article, we’ll walk through how to set up a database on Render and copy the URL into your .env file.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
import os
from dotenv import load_dotenv
from utils.db_handler import DatabaseHandler
import pandas as pd
import uuid
import sys
from sqlalchemy.exc import OperationalError
import psycopg2

# Loading environmental variables
load_dotenv(override=True)

# Construct PostgreSQL connection URL for Render
URL_database = os.environ.get("External_Database_Url")

# Initialize DatabaseHandler with our URL
engine = DatabaseHandler(URL_database)

# loading initial user data
users_df = pd.read_csv("Data/users.csv")
games_df = pd.read_csv("Data/games.csv")
user_games_df = pd.read_csv("Data/user_games.csv")
user_recommendations_df = pd.read_csv("Data/user_recommendations.csv")
game_tags_df = pd.read_csv("Data/game_tags.csv")

First, we load five CSV files into dataframes from our Data folder; we have one file for each of the tables shown in our database diagram. We also establish a connection to our data by declaring an engine variable; this engine variable uses a custom DataBaseHandler class with the initialization method shown below. This class takes a connection string to our database on Render(or your preferred cloud service), passed in from our .env file, and contains all of our database connect, update, delete, and test functionalities.

After loading our data and instantiating our DatabaseHandler class, we then have to define a query to create each of the five tables and execute these queries using the DatabaseHandler.create_table function. This is a very simple function that connects to our database, executes the query, and closes the connection, leaving us with the five tables shown in our database diagram; however, they are currently empty.

# Defining queries to create tables
user_table_creation_query = """CREATE TABLE IF NOT EXISTS users (
    id UUID PRIMARY KEY,
    username VARCHAR(255) UNIQUE NOT NULL,
    password VARCHAR(255) NOT NULL,
    email VARCHAR(255) NOT NULL,
    role VARCHAR(50) NOT NULL
    )
    """
game_table_creation_query = """CREATE TABLE IF NOT EXISTS games (
    id UUID PRIMARY KEY,
    appid VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    type VARCHAR(255),
    is_free BOOLEAN DEFAULT FALSE,
    short_description TEXT,
    detailed_description TEXT,
    developers VARCHAR(255),
    publishers VARCHAR(255),
    price VARCHAR(255),
    genres VARCHAR(255),
    categories VARCHAR(255),
    release_date VARCHAR(255),
    platforms TEXT,
    metacritic_score FLOAT,
    recommendations INTEGER
    )
    """

user_games_query = """CREATE TABLE IF NOT EXISTS user_games (
    id UUID PRIMARY KEY,
    username VARCHAR(255) NOT NULL,
    appid VARCHAR(255) NOT NULL,
    shelf VARCHAR(50) DEFAULT 'Wish_List',
    rating FLOAT DEFAULT 0.0,
    review TEXT
    )
    """
recommendation_table_creation_query = """CREATE TABLE IF NOT EXISTS user_recommendations (
    id UUID PRIMARY KEY,
    username VARCHAR(255),
    appid VARCHAR(255),
    similarity FLOAT
    )
    """

game_tags_creation_query = """CREATE TABLE IF NOT EXISTS game_tags (
    id UUID PRIMARY KEY,
    appid VARCHAR(255) NOT NULL,
    category VARCHAR(255) NOT NULL
    )
    """



# Running queries to create tables
engine.delete_table('user_recommendations')
engine.delete_table('user_games')
engine.delete_table('game_tags')
engine.delete_table('games')
engine.delete_table('users')

# Create tables
engine.create_table(user_table_creation_query)
engine.create_table(game_table_creation_query)
engine.create_table(user_games_query)
engine.create_table(recommendation_table_creation_query)
engine.create_table(game_tags_creation_query)

Following the initial table setup, we then run a quality check to ensure each of our datasets has the required ID column, populate the data from the dataframes into the appropriate table, and then test to ensure that the tables were populated correctly. The test_table function returns a dictionary that will be of the form {‘table_exists’: True, ‘table_has_data’: True} if the setup worked correctly.

# Ensuring each row of each dataframe has a unique ID
if 'id' not in users_df.columns:
    users_df['id'] = [str(uuid.uuid4()) for _ in range(len(users_df))]
if 'id' not in games_df.columns:
    games_df['id'] = [str(uuid.uuid4()) for _ in range(len(games_df))]
if 'id' not in user_games_df.columns:
    user_games_df['id'] = [str(uuid.uuid4()) for _ in range(len(user_games_df))]
if 'id' not in user_recommendations_df.columns:
    user_recommendations_df['id'] = [str(uuid.uuid4()) for _ in range(len(user_recommendations_df))]
if 'id' not in game_tags_df.columns:
    game_tags_df['id'] = [str(uuid.uuid4()) for _ in range(len(game_tags_df))]

# Populates the 4 tables with data from the dataframes
engine.populate_table_dynamic(users_df, 'users')
engine.populate_table_dynamic(games_df, 'games')
engine.populate_table_dynamic(user_games_df, 'user_games')
engine.populate_table_dynamic(user_recommendations_df, 'user_recommendations')
engine.populate_table_dynamic(game_tags_df, 'game_tags')

# Testing if the tables were created and populated correctly
print(engine.test_table('users'))
print(engine.test_table('games'))
print(engine.test_table('user_games'))
print(engine.test_table('user_recommendations'))
print(engine.test_table('game_tags'))

Getting Started with FastAPI

Now that we have our database set up and populated, we need to build the methods to access, update, and delete data, using FastAPI. FastAPI enables us to easily build standardized(and fast) API’s to enable interaction with our database. The FastAPI docs offer a great step-by-step tutorial that can be found here. As a high-level summary, there are several great features that make FastAPI ideal for serving as the interaction layer between a database and a front-end application.

1. Standardization: FastAPI allows us to define routes to interact with our tables in a standardized way using GET, POST, DELETE, UPDATE, etc. methods. This standardization enables us to build a data access layer in pure Python that can then be interact with a wide variety of front-end application. We simply call the API methods we want in the front end, regardless of what language its constructed in.
2. Data Validation: As we’ll show below, we need to define a Pydantic data model for each object we interact with(think our games and user tables). The main advantage of this is that it ensures all our variables have defined data types, for example, if we define our Game object such that the rating field is of type float and a user tries to make a post request to add a new entry with a rating of “great” it wont work. This built-in data validation will help us prevent all sorts of data quality issues from as our system scales.
3. Asynchronous: FastAPI functions can run asynchronously, meaning one of them isn’t dependent on the other finishing. This can significantly improve the performance because we won’t have a slow Fast task waiting on a slow one to complete.
4. Swagger Docs Built In: FastAPI has a built-in UI that we can navigate to on localhost, enabling us to easily test and interact with our routes.

FastAPI Models

The FastAPI portion of our project relies on two main files: models.py, which defines the data models that we’ll be interacting with (games, users, etc.), and main.py, which defines our actual FastAPI App and contains our routes. In the context of FastAPI, Routes define the different paths for processing requests. For example, we might have a /games route to request games from our database.

First, let’s discuss our models.py file. In this file, we define all of our models. While we have different models for different objects the general approach will be the same so we’ll only discuss the games model, shown below, in detail. The first thing you’ll notice below is that we have two actual classes defined for our Game object: a GameModel class that inherits from the Pydantic base model, and a Game class that inherits from the sqlalchemy declarative_base. The natural question then is, why do we have two classes for one data structure(our game’s data structure)?

If we weren’t using an SQL database for this project and instead read each of our CSV files into a dataframe every time main.py was run, then we wouldn’t need the Game class, only the GameModel class. In this scenario, we would read in our games.csv dataframe, and FastAPI would use the GameModel class to ensure datatypes were correctly adhered to.

However, because we are using an SQL database, it makes more sense to have separate classes for our API and our database, as the two classes have slightly different jobs. Our API class handles data validation, serialization, and optional fields, and our database class handles database-specific concerns like defining primary/foreign keys, defining which table the object maps to, and protecting secure data. To reiterate the last point, we might have sensitive fields in our database that are for internal consumption only, and we don’t want to expose them to a user through an API( password for example). We can address this concern by having a separate user-facing Pydantic class and an Internal SQL Alchemy one.

Below is an example of how this can be implemented for our Games object; we have separate classes defined for our other tables, which can be found here; however, the general structure is the same.

from pydantic import BaseModel
from uuid import UUID,uuid4
from typing import Optional
from enum import Enum
from sqlalchemy import Column, String, Float, Integer
import sqlalchemy.dialects.postgresql as pg
from sqlalchemy.dialects.postgresql import UUID as SA_UUID
from sqlalchemy.ext.declarative import declarative_base
import uuid
from uuid import UUID

# loading sql model
from sqlmodel import Field, Session, SQLModel, create_engine, select

# Initialize the base class for SQLAlchemy models
Base = declarative_base()

# This is the Game model for the database
class Game(Base):
    __tablename__ = "optigame_products"  # Table name in the PostgreSQL database

    id = Column(pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, unique=True, nullable=False)
    appid = Column(String, unique=True, nullable=False)  
    name = Column(String, nullable=False)  
    type = Column(String, nullable=True)  
    is_free = Column(pg.BOOLEAN, nullable=True, default=False)  #
    short_description = Column(String, nullable=True)  
    detailed_description = Column(String, nullable=True)  
    developers = Column(String, nullable=True)  
    publishers = Column(String, nullable=True)  
    price = Column(String, nullable=True)  
    genres = Column(String, nullable=True)  
    categories = Column(String, nullable=True)  
    release_date = Column(String, nullable=True)  
    platforms = Column(String, nullable=True)  
    metacritic_score = Column(Float, nullable=True)  
    recommendations = Column(Integer, nullable=True)  

class GameModel(BaseModel):
    id: Optional[UUID] = None
    appid: str
    name: str
    type: Optional[str] = None
    is_free: Optional[bool] = False
    short_description: Optional[str] = None
    detailed_description: Optional[str] = None
    developers: Optional[str] = None
    publishers: Optional[str] = None
    price: Optional[str] = None
    genres: Optional[str] = None
    categories: Optional[str] = None
    release_date: Optional[str] = None
    platforms: Optional[str] = None
    metacritic_score: Optional[float] = None
    recommendations: Optional[int] = None

    class Config:
        orm_mode = True  # Enable ORM mode to work with SQLAlchemy objects
        from_attributes = True # Enable attribute access for SQLAlchemy objects

Setting Up FastAPI Routes

After we have our Models defined, we can then create methods to interact with these models and request data from the database(GET), add data to the Database(POST), or remove data from the database(DELETE). Below is an example of how we can define a GET request for our games model. We have some initial setup at the beginning of our main.py function to fetch the database URL and connect to it. Then we initialize our app and add middleware to define which URLs we’ll accept requests from. Because we’ll be deploying the FastAPI project on Render and sending requests to it from our local machine, the only origin we’re allowing is localhost port 8000. We then define our app.get method called fetch_products, which takes an appid input, queries our database for Game objects where appid is equal to our filtered appid, and returns these products.

Note the below snipped contains just the setup and first get method, the rest are fairly similar and available on the Repo, so we won’t give an in-depth explanation for each one here.

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from uuid import uuid4, UUID
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()

# security imports
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer

# custom imports
from src.models import User, Game, GameModel, UserModel,  UserGameModel, UserGame, GameSimilarity,GameSimilarityModel, UserRecommendation, UserRecommendationModel
from src.similarity_pipeline import UserRecommendationService

# Load the database connection string from environment variable or .env file
DATABASE_URL = os.environ.get("Internal_Database_Url")

# creating connection to the database
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Create the database tables (if they don't already exist)
Base.metadata.create_all(bind=engine)

# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Initialize the FastAPI app
app = FastAPI(title="Game Store API", version="1.0.0")

# Add CORS middleware to allow requests 
origins = ["http://localhost:8000"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,  
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)



#-------------------------------------------------#
# ----------PART 1: GET METHODS-------------------#
#-------------------------------------------------#
@app.get("/")
async def root():
    return {"message": "Hello World"}


@app.get("/api/v1/games/")
async def fetch_products(appid: str = None, db: Session = Depends(get_db)):
    # Query the database using the SQLAlchemy Game model
    if appid:
        products = db.query(Game).filter(Game.appid == appid).all()
    else:
        products = db.query(Game).all()
    return [GameModel.from_orm(product) for product in products]

Once we have our main.py defined, we can finally run it from our base project directory using the command below.

uvicorn src.main:app --reload

Once this is done, we can navigate to http://127.0.0.1:8000/docs and see the below interactive FastAPI environment. From this page, we can test any of our methods defined in our main.py file. In the case of our fetch_products function, we can pass it an appid and return any matching games from our database.

Building our Similarity Pipeline

We have our database set up and can access and update data via FastAPI; it’s now time to turn to the central feature of this project: a recommender pipeline. Recommender systems are a well-researched field, and we’re not adding any innovation here; however, this will offer a clear example of how to implement a basic recommender system using FastAPI.

Getting Started — How to Recommend Products?

If we think about the question “How would I recommend new products that a user will like?”, there are two approaches that make intuitive sense.

1. Collaborative Recommendation Systems: If I have a series of users and a series of products, I could identify users with similar interests by looking at their overall basket of products and then identify products ‘missing’ from a given user’s basket. For example, if I have users 1–3 and products A-C, users 1–2 like all three products, but user 3 has thus far only liked products A + B, then I might recommend them product C. This logically makes sense; all three users have a high degree of overlap in products that they’ve liked, but product C is missing from user 3’s basket, there’s a high likelihood that they would like it as well. This process of generating recommendations by comparing like users is called collaborative filtering.
2. Content-Based Recommendation System: If I have a series of products, I could identify products that are similar to products that a user has liked and recommend those products. For example, if I have a series of tags for each game, I could convert each game’s series of tags into a vector of 1s and 0s and then use a similarity measure (in this case, a cosine similarity measure) to measure the similarity between games based on their vectors. Once I have done this, I can then return the top N most similar games to those liked by a user based on their similarity score.

More on Recommender Systems can be found here.

Because our initial dataset doesn’t have a large volume of users, we don’t have the necessary data to suggest items based on user similarity, which is known as a cold start problem. As a result, we will instead develop a content-based recommender system as we have a significant amount of game data to work with.

To build our pipeline, we have to address two challenges: (1) How do we go about calculating similarity scores for a user, and (2) how do we automate this to run whenever a user makes an update to their games?

We’ll go over how a similarity pipeline can be triggered each time a user makes a POST request by ‘liking’ a game, and then cover how to build the pipeline itself.

Tying Recommender Pipeline to FastAPI

For now, imagine we have a Recommendation Service that will update our user_recommendation table. We want to ensure that this service is called whenever a user updates their preferences. We can implement this in a couple of steps as shown below; first, we define a generate_recommendations_background function, this function is responsible for connecting to our database, running the similarity pipeline, and then closing the connection. Next, we need to ensure this is called when a user makes a post request(i.e., likes a new game); to do this, we simply add the function call at the end of our create_user_game post request function.

The result of this workflow is that whenever a user makes a post request to our user_game table, they call the create_user_game function, add a new user_game object to the database, and then run the similarity pipeline as a background function.

Note: The Below post method and helper function are stored in main.py with the rest of our FastAPI methods.

# importing similarity pipeline
from src.similarity_pipeline import UserRecommendationService

# Background task function
def generate_recommendations_background(username: str, database_url: str):
    """Background task to generate recommendations for a user"""
    # Create a new database session for the background task
    background_engine = create_engine(database_url)
    BackgroundSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=background_engine)
    
    db = BackgroundSessionLocal()
    try:
        recommendation_service = UserRecommendationService(db, database_url)
        recommendation_service.generate_recommendations_for_user(username)
    finally:
        db.close()

# Post method which calls background task function
@app.post("/api/v1/user_game/")
async def create_user_game(user_game: UserGameModel, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    # Check if the entry already exists
    existing = db.query(UserGame).filter_by(username=user_game.username, appid=user_game.appid).first()
    if existing:
        raise HTTPException(status_code=400, detail="User already has this game.")

    # Prepare data with defaults
    user_game_data = {
        "username": user_game.username,
        "appid": user_game.appid,
        "shelf": user_game.shelf if user_game.shelf is not None else "Wish_List",
        "rating": user_game.rating if user_game.rating is not None else 0.0,
        "review": user_game.review if user_game.review is not None else ""
    }
    if user_game.id is not None:
        user_game_data["id"] = UUID(str(user_game.id))

    # Save the user game to database
    db_user_game = UserGame(**user_game_data)
    db.add(db_user_game)
    db.commit()
    db.refresh(db_user_game)
    
    # Trigger background task to generate recommendations for this user
    background_tasks.add_task(generate_recommendations_background, user_game.username, DATABASE_URL)
    
    return db_user_game

Building Recommender Pipeline

Now that we understand how our similarity pipeline can be triggered when a user updates their liked games, it’s time to dive into the mechanics of how the recommendation pipeline works. Our recommendation pipeline is stored in similarity_pipeline.py and contains our UserRecommendationService class that we showed how to import and instantiate above. This class contains a series of helper functions that are ultimately all called in the generate_recommendations_for_user method. There are 7 basic steps called in order that we’ll walk through one by one.

1. Fetching a user’s Games: To generate similar game recommendations, we need to retrieve the games that a user has already added to their game basket. This is done by calling our fetch_user_games helper function. This function queries our user_games table using the user’s ID, which is making the post request as an input, and returning all games in their basket.
2. Fetching game tags: To compare games, we need a dimension to compare them on, and that dimension is the tags associated with each game(strategy, board game, etc.). To retrieve the game:tag mapping, we call our fetch_all_game_tags function, which returns the tags for all the games in our database
3. Vectorizing game tags: To compare the similarity between games A and B, we first need to vectorize the game tags using our create_game_vectors function. This function takes a series of all tags in alphabetical order and checks if each of the tags is associated with a given game. For example, if our total set of tags was [boardgame, deckbuilding, resource-management] and game 1 just had the boardgame tag associated with it, then its vector would be [1, 0, 0].
4. Creating our users vector: once we have a vector representing each game, we then need an aggregate user vector to compare it to. To achieve this, we use our create_user_vector function, which generates an aggregate vector of the same length as our game vectors that we can then use to generate a similarity score between our user and every other game.
5. Calculate Similarity: We use the vectors created in steps 3 and 4 in our calculate_user_recommendations, which calculates a cosine similarity score ranging from 0–1 and measuring the similarity between each game and our user aggregate games
6. Deleting old Recommendations: Before we populate our user_recommendations table with new recommendations for a user, we first have to delete the old ones with delete_existing_recommendations. This deletes just the recommendations for the user who made the post request; the others remain the same.
7. Populate new Recommendations: After deleting the old recommendations, we then populate the new ones with save_recommendations.

from sqlalchemy.orm import Session
from sqlalchemy import create_engine, text
from src.models import UserGame, UserRecommendation
from sklearn.metrics.pairwise import cosine_similarity
import pandas as pd
import uuid
from typing import List
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class UserRecommendationService:
    def __init__(self, db_session: Session, database_url: str):
        self.db = db_session
        self.database_url = database_url
        self.engine = create_engine(database_url)

    def fetch_user_games(self, username: str) -> pd.DataFrame:
        """Fetch all games for a specific user"""
        query = text("SELECT username, appid FROM user_games WHERE username = :username")
        with self.engine.connect() as conn:
            result = conn.execute(query, {"username": username})
            data = result.fetchall()
            return pd.DataFrame(data, columns=['username', 'appid'])

    def fetch_all_category(self) -> pd.DataFrame:
        """Fetch all game tags"""
        query = text("SELECT appid, category FROM category")
        with self.engine.connect() as conn:
            result = conn.execute(query)
            data = result.fetchall()
            return pd.DataFrame(data, columns=['appid', 'category'])

    def create_game_vectors(self, tag_df: pd.DataFrame) -> tuple[pd.DataFrame, List[str], List[str]]:
        """Create game vectors from tags"""
        unique_tags = tag_df['category'].drop_duplicates().sort_values().tolist()
        unique_games = tag_df['appid'].drop_duplicates().sort_values().tolist()
        
        game_vectors = []
        for game in unique_games:
            tags = tag_df[tag_df['appid'] == game]['category'].tolist()
            vector = [1 if tag in tags else 0 for tag in unique_tags]
            game_vectors.append(vector)
        
        return pd.DataFrame(game_vectors, columns=unique_tags, index=unique_games), unique_tags, unique_games

    def create_user_vector(self, user_games_df: pd.DataFrame, game_vectors: pd.DataFrame, unique_tags: List[str]) -> pd.DataFrame:
        """Create user vector from their played games"""
        if user_games_df.empty:
            return pd.DataFrame([[0] * len(unique_tags)], columns=unique_tags, index=['unknown_user'])
        
        username = user_games_df.iloc[0]['username']
        user_games = user_games_df['appid'].tolist()
        
        # Only keep games that exist in game_vectors
        user_games = [g for g in user_games if g in game_vectors.index]
        
        if not user_games:
            user_vector = [0] * len(unique_tags)
        else:
            played_game_vectors = game_vectors.loc[user_games]
            user_vector = played_game_vectors.mean(axis=0).tolist()
        
        return pd.DataFrame([user_vector], columns=unique_tags, index=[username])

    def calculate_user_recommendations(self, user_vector: pd.DataFrame, game_vectors: pd.DataFrame, top_n: int = 20) -> pd.DataFrame:
        """Calculate similarity between user vector and all game vectors"""
        username = user_vector.index[0]
        user_vector_data = user_vector.iloc[0].values.reshape(1, -1)
        
        # Calculate similarities
        similarities = cosine_similarity(user_vector_data, game_vectors)
        similarity_df = pd.DataFrame(similarities.T, index=game_vectors.index, columns=[username])
        
        # Get top N recommendations
        top_games = similarity_df[username].nlargest(top_n)
        
        recommendations = []
        for appid, similarity in top_games.items():
            recommendations.append({
                "username": username,
                "appid": appid,
                "similarity": float(similarity)
            })
        
        return pd.DataFrame(recommendations)

    def delete_existing_recommendations(self, username: str):
        """Delete existing recommendations for a user"""
        self.db.query(UserRecommendation).filter(UserRecommendation.username == username).delete()
        self.db.commit()

    def save_recommendations(self, recommendations_df: pd.DataFrame):
        """Save new recommendations to database"""
        for _, row in recommendations_df.iterrows():
            recommendation = UserRecommendation(
                id=uuid.uuid4(),
                username=row['username'],
                appid=row['appid'],
                similarity=row['similarity']
            )
            self.db.add(recommendation)
        self.db.commit()

    def generate_recommendations_for_user(self, username: str, top_n: int = 20):
        """Main method to generate recommendations for a specific user"""
        try:
            logger.info(f"Starting recommendation generation for user: {username}")
            
            # 1. Fetch user's games
            user_games_df = self.fetch_user_games(username)
            if user_games_df.empty:
                logger.warning(f"No games found for user: {username}")
                return
            
            # 2. Fetch all game tags
            tag_df = self.fetch_all_category()
            if tag_df.empty:
                logger.error("No game tags found in database")
                return
            
            # 3. Create game vectors
            game_vectors, unique_tags, unique_games = self.create_game_vectors(tag_df)
            
            # 4. Create user vector
            user_vector = self.create_user_vector(user_games_df, game_vectors, unique_tags)
            
            # 5. Calculate recommendations
            recommendations_df = self.calculate_user_recommendations(user_vector, game_vectors, top_n)
            
            # 6. Delete existing recommendations
            self.delete_existing_recommendations(username)
            
            # 7. Save new recommendations
            self.save_recommendations(recommendations_df)
            
            logger.info(f"Successfully generated {len(recommendations_df)} recommendations for user: {username}")
            
        except Exception as e:
            logger.error(f"Error generating recommendations for user {username}: {str(e)}")
            self.db.rollback()
            raise

Wrapping Up

In this article, we covered how to set up a PostgreSQL database and FastAPI application to run a game recommender system. However, we haven’t yet gone over how to deploy this system to a cloud service to allow others to interact with it. For part two covering exactly this, read on in Part 2.

Figures: All images, unless otherwise noted, are by the author.

Links

1. Github Repository for Project: https://github.com/pinstripezebra/recommender_system
2. FastAPI Docs: https://fastapi.tiangolo.com/tutorial/
3. Recommender Systems: https://en.wikipedia.org/wiki/Recommender_system
4. Cosine Similarity: https://en.wikipedia.org/wiki/Cosine_similarity)