使用 FastAPI、PostgreSQL 和 Render 部署视频游戏推荐系统：实战篇二

项目启动与概述

在系列文章的第一部分中，详细介绍了如何利用 FastAPI 和 PostgreSQL 搭建一个桌面游戏推荐系统。本文作为系列的第二部分，将深入探讨如何将该项目部署到云服务平台（以 Render 为例），从而实现其广泛的用户可访问性与可用性。

为实现这一目标，本部分内容将详细阐述以下核心环节：在 Render 上配置 PostgreSQL 数据库并导入数据；对 FastAPI 应用程序进行 Docker 容器化；最终将其部署为一个可公开访问的 Render Web 应用程序。

目录概览

1. 在 Render 上部署 PostgreSQL 数据库
2. 将 FastAPI 应用作为 Render Web 应用程序进行部署

– 应用程序的 Docker 容器化

– 将 Docker 镜像推送到 DockerHub

– 从 DockerHub 拉取镜像到 Render

所用工具

• Render
• Docker Desktop
• Docker Hub

在 Render 上进行部署

目前，本地环境中已搭建并成功运行 PostgreSQL 数据库和 FastAPI 应用程序。接下来，将它们部署到云服务平台，以便前端应用或最终用户（通过 Swagger UI）能够访问。本项目选择使用 Render，Render 作为一款云平台，为小型项目的部署提供了比AWS和Azure等大型云服务提供商更为简洁高效的设置体验。

开始部署前，请访问 Render 官网并创建一个新账户。随后，通过点击页面上方的“New Project”（新建项目）按钮即可创建一个新项目。请注意，截至本文撰写时，Render 提供了一个试用期，允许用户在首月免费体验。项目将命名为“fastapi-test”，创建成功后即可进入该项目。

图 2：Render 中的新项目

每个项目都包含其独立运行所需的所有组件。对于本项目而言，主要需要两个核心组件：一个数据库和一个用于 FastAPI 应用程序的 Web 服务器。首先从创建数据库开始。

图 3：FastAPI 项目

操作非常简单，如图 3 所示，选择“Create New Service”（创建新服务），然后选择“Postgres”。接着，将进入图 4 所示的界面来配置数据库。将数据库命名为“fastapi-database”，并选择免费套餐以启动初步体验。Render 的免费数据库套餐有使用时间限制，但对于本示例来说已足够。如果需要长期维护数据库，其收费套餐价格也相当合理。

图 4：Render 上的 Postgres 设置

输入数据库信息并选择“Create”（创建）后，数据库的设置将耗时约一分钟。完成后，将看到图 5 所示的界面。此时，务必将“Internal Database URL”和“External Database URL”这两个环境变量保存到本地的 .env 文件中，它们是 FastAPI 应用程序连接数据库的关键。随后，可以使用“External Database URL”变量（因为它允许从 Render 环境之外的本地机器连接）测试数据库连接，并在本地机器上创建表结构，然后继续配置 FastAPI 应用程序。

图 5：Render 数据库凭据

接下来，运行数据库连接测试脚本。该脚本尝试使用 External_Database_Url 变量作为连接字符串来连接数据库，并创建一个测试表。值得注意的是，External_Database_Url 是数据库的完整连接字符串，因此可以直接将其作为单个输入参数传递。成功运行后，将显示如图 6 所示的输出。

图 6：从本地机器成功连接到数据库

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

# Load environment variables from .env file (override=True reloads changed values)
load_dotenv(override=True)
# loaidng external database URL
database_url = os.environ.get("External_Database_Url")
if not database_url:
    print(" External_Database_Url not found in environment variables")
    print("Please check your .env file contains: External_Database_Url=your_render_postgres_url")
    sys.exit(1)
print(f"Database URL loaded: {database_url[:50]}...")
# Parse the database URL to extract components for testing
from urllib.parse import urlparse
import socket
def parse_database_url(url):
    """Parse database URL to extract connection components"""
    parsed = urlparse(url)
    return {
        'host': parsed.hostname,
        'port': parsed.port or 5432,
        'database': parsed.path.lstrip('/'),
        'username': parsed.username,
        'password': parsed.password
    }
db_params = parse_database_url(database_url)
def test_network_connectivity():
    """Test network connectivity to Render PostgreSQL endpoint"""
    print("
=== Network Connectivity Tests ===")
    # 1. Test DNS resolution
    try:
        ip_address = socket.gethostbyname(db_params['host'])
        print(f" DNS Resolution successful")
    except socket.gaierror as e:
        print(f" DNS Resolution failed: {e}")
        return False

    # 2. Test port connectivity
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)  # 10 second timeout
        result = sock.connect_ex((db_params['host'], int(db_params['port'])))
        sock.close()

        if result == 0:
            print(f" Port {db_params['port']} is accessible")
            return True
        else:
            print(f" Port {db_params['port']} is NOT accessible")
            print("   This might indicate a network connectivity issue")
            return False
    except Exception as e:
        print(f" Port connectivity test failed: {e}")
        return False
# Run connectivity tests
network_ok = test_network_connectivity()
if not network_ok:
    print("
 TROUBLESHOOTING STEPS:")
    print("1. Check your internet connection")
    print("2. Verify the Render PostgreSQL URL is correct")
    print("3. Ensure your Render PostgreSQL instance is active")
    print("4. Check if there are any Render service outages")
    sys.exit(1)
print("
=== Attempting Database Connection ===")
# connect to the database using psycopg2
try:
    conn = psycopg2.connect(
            host=db_params['host'],
            database=db_params['database'],
            user=db_params['username'],
            password=db_params['password'],
            port=db_params['port'],
            connect_timeout=30  # 30 second timeout
        )

    # If the connection is successful, you can perform database operations
    cursor = conn.cursor()

    # Example: Execute a simple query
    cursor.execute("SELECT version();")
    db_version = cursor.fetchone()
    print(f" PostgreSQL Database Version: {db_version[0]}")

    # Test creating a simple table to verify permissions
    cursor.execute("CREATE TABLE IF NOT EXISTS connection_test (id SERIAL PRIMARY KEY, test_time TIMESTAMP DEFAULT NOW());")
    conn.commit()
    print(" Database permissions verified - can create tables")

    cursor.close()
    conn.close()
    print(" psycopg2 connection successful!")

except psycopg2.OperationalError as e:
    print(f" Database connection failed: {e}")
    if "timeout" in str(e).lower():
        print("
 TIMEOUT TROUBLESHOOTING:
")
        print("- Check your internet connection")
        print("- Verify the Render PostgreSQL URL is correct")
        print("- Check if Render service is experiencing issues")
    elif "authentication" in str(e).lower():
        print("
 AUTHENTICATION TROUBLESHOOTING:
")
        print("- Verify the database URL contains correct credentials")
        print("- Check if your Render PostgreSQL service is active")
        print("- Ensure the database URL hasn't expired or changed")
    sys.exit(1)
except Exception as e:
    print(f" Unexpected error: {e}")
    sys.exit(1)
# If we get here, connection was successful, so exit the test
print(f"
 All tests passed! Render PostgreSQL connection is working.")
print(f" Connected to database: {db_params['database']}")
print(" Ready for use in your application!")

数据库数据加载

既然已确认可以从本地机器成功连接到 Render 上的数据库，下一步是配置数据库表结构并填充数据。为了加载数据库，将使用 src/load_database.py 文件。该脚本的各个组成部分已在系列文章的第一部分中进行过详细介绍，因此在本部分不再深入探讨其内部细节。值得强调的是，此处再次使用 External_Database_Url 作为连接字符串。此外，在脚本末尾，调用了 DatabaseHandler 类中定义的 test_table 函数。此函数会尝试连接到传入的表名，并返回该表中的行数，以验证数据是否成功加载。

运行此脚本后，将获得如图 11 所示的输出。图中显示了所有表均已成功创建，并且在最后重新检查了是否可以从这些表中返回数据，并验证了输出行数与输入行数一致。

图 7：数据库成功填充数据

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

# Load environment variables from .env file
load_dotenv(override=True)

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

# Initialize DatabaseHandler with the constructed URL
engine = DatabaseHandler(URL_database)

# loading initial user data
users_df = pd.read_csv("Data/steam_users.csv")
games_df = pd.read_csv("Data/steam_games.csv")
user_games_df = pd.read_csv("Data/steam_user_games.csv")
user_recommendations_df = pd.read_csv("Data/user_recommendations.csv")
game_tags_df = pd.read_csv("Data/steam_game_tags.csv")

# 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)

# 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'))

在 Render 上部署 FastAPI 应用程序

项目的数据库部分已成功部署到 Render，现在是时候配置并部署 FastAPI 应用程序了。为此，将利用 Render 提供的 Web 应用程序托管服务，将 FastAPI 应用作为一个可被外部服务访问的 Web 应用程序进行部署。如果目标是构建一个全栈应用，前端可以向 Render 上的 FastAPI 应用程序发送请求并获取数据。然而，由于当前阶段不涉及前端组件的开发，将通过 Swagger UI 来与应用程序进行交互和测试。

使用 Docker 对应用程序进行容器化

FastAPI 项目已在本地环境中成功配置，但现在需要将其所有代码、依赖项和环境变量一并迁移到 Render 上的一个容器中。这可能是一项艰巨的任务。幸运的是，Docker 能够优雅地处理所有复杂的容器化环节，极大简化了开发者的工作，通过一个简单的配置文件和几条命令即可实现这一目标。对于尚未接触过 Docker 的读者，可以参考此处的优秀教程。简而言之，Docker 是一款通过将应用程序及其所有依赖项打包成一个镜像，然后将该镜像部署到 Render 等服务，从而简化应用程序部署和管理流程的工具。本项目将 DockerHub 作为镜像仓库，它是一个集中式、版本受控的镜像存储区域，Render 可以从中拉取镜像。

本项目的工作流程可概括为：在本地运行的 FastAPI 应用 → 使用 Docker 创建“快照”并存储为 Docker 镜像 → 将该镜像推送到 DockerHub → Render 从 DockerHub 拉取此镜像，并用其在 Render 服务器上启动一个运行应用程序的容器。开始这一过程（接下来将详细介绍）需要先安装 Docker Desktop。Docker 的安装过程非常直接，可在此处开始：https://www.docker.com/products/docker-desktop/

此外，如果尚未拥有 Docker Hub 账户，则需要创建一个。Docker Hub 将作为存储 Docker 镜像并将其拉取到 Render 的仓库。您可以在此处创建 Docker Hub 账户：https://hub.docker.com/。

构建 Docker 镜像

为了为项目创建 Docker 镜像，首先要确保 Docker Desktop 正在运行；如果未运行，尝试创建 Docker 镜像时很可能会遇到错误。为确认其运行状态，请从搜索栏或桌面打开 Docker Desktop 应用程序，点击左下角的三个点，并确保看到绿点以及“Docker Desktop is running”的提示。

图 8：Docker Desktop 正在运行

接下来，需要通过定义 Dockerfile 来指示 Docker 如何构建镜像。Dockerfile 如图 9 所示，它保存在项目的顶层目录中，并提供了指导 Docker 将应用程序打包成可在不同硬件上部署的镜像的指令。下面将逐一解析该文件，以理解其作用。

1. FROM：选择基础镜像： Dockerfile 的第一行指定了作为应用程序扩展基础的基础镜像。此处使用的是 python:3.13-slim-bullseye 镜像，它是一个轻量级的基于 Debian 的镜像，将作为应用程序的运行环境。
2. WORKDIR：更改工作目录： 这一步将容器内部的默认工作目录设置为 /app。
3. RUN：检查系统依赖项更新。
4. COPY：复制 requirements.txt 文件。确保 requirements.txt 是最新且包含项目所需的所有库至关重要，否则镜像在启动时将无法正确运行。
5. RUN：安装 requirements.txt 文件中列出的依赖项。
6. COPY：将本地目录中的整个项目复制到 /app 目录（在步骤 2 中创建）。
7. RUN：在 /app 目录下创建 logs 目录。
8. EXPOSE：声明将要暴露的端口为 8000。
9. ENV：将 Python 路径设置为 /app。
10. CMD：使用 Uvicorn 运行 FastAPI 应用程序，将应用程序指向 src.main:app 中定义的实例，并在 8000 端口上运行。

图 9：Dockerfile 文件

定义 Dockerfile 后，就拥有了一组指令，可以将其提供给 Docker 以容器化应用程序为镜像，然后推送到 Docker Hub。此操作可通过在 VS Code 终端中，从项目根目录执行以下几条命令来完成，这些命令将引导 Docker 完成镜像的构建、标签设置及推送至 Docker Hub 的过程。

1. 首先，构建 Docker 镜像，这可能需要一两分钟。在此示例中，镜像命名为“recommendersystem”。
2. 接下来，为镜像打标签。语法为 image_name user_name/docker_hub_folder:image_name_on_dockerhub。
3. 最后，再次指定 user_name/docker_hub_folder:image_name_on_dockerhub，将镜像推送到 DockerHub。

docker build -t recommendersystem .
docker tag recommendersystem seelucas/fastapi_tutorial:fastapi_on_render
docker push seelucas/fastapi_tutorial:fastapi_on_render

完成这些步骤后，应该能够登录 DockerHub，导航到项目，并看到一个名称与之前三条命令中给定的名称（在本例中为 fastapi_on_render）匹配的镜像。

图 10：Dockerhub 上的 Docker 镜像

将 Docker 镜像拉取到 Render

Docker 镜像已成功上传至 DockerHub，现在是时候将其部署到 Render 上了。此操作通过导航到之前创建数据库的同一项目“fastapi-test”来完成。在右上角选择“New”（新建），然后选择“Web Service”（Web 服务），因为 FastAPI 应用程序将作为 Web 应用进行部署。

由于镜像将从 DockerHub 部署，因此需要指定源代码为“Existing Image”（现有镜像）。如图 11 所示，将希望部署的镜像在 DockerHub 上的目录路径粘贴到 Render 的“Image URL”（镜像 URL）字段中。此时，系统将提示这是一个私有镜像，这意味着需要创建一个 DockerHub 访问令牌，以便安全地将镜像从 DockerHub 拉取到 Render。

图 11：Render 上的 Docker 镜像

创建 DockerHub 访问令牌的过程非常直接：导航到 DockerHub 账户 → 设置 → 个人访问令牌（Personal Access token）。屏幕将显示如图 12 所示的界面。在此处提供访问令牌的名称、过期日期和权限。由于只是将镜像拉取到 Render，因此仅需要读取权限，而非写入或删除权限，故选择读取权限。

图 12：为 Dockerhub 创建个人访问令牌

最后，选择“Generate”（生成）将生成令牌，然后需要将其复制并粘贴到 Render 中，如图 13 所示。

图 13：Render 上的 Docker 凭据

点击“Add Credential”（添加凭据）后，系统将加载一分钟以保存凭据。随后，将返回到上一屏幕，在此处可以选择用于连接 DockerHub 的凭据。在此案例中，将选择并使用刚才创建的教程凭据，随后点击“Connect”按钮建立连接。至此，已成功建立了从 DockerHub 拉取 Docker 镜像到 Render 进行部署的安全连接。

图 14：带有有效凭据的 Render Web 服务

在下一页，继续设置 Render Web 应用程序，首先选择免费选项。重要的是，在环境变量（Environmental Variables）部分，复制并粘贴 .env 文件中的内容。尽管并非所有变量都使用，但其中包含的“Internal_Database_Url”是 FastAPI 在 main.py 文件中查找的数据库 URL。若缺少此变量，应用程序将无法正常连接到数据库，因此其提供是绝对关键且不可或缺的。请注意：为了测试目的，之前使用的是“External_Database_Url”，因为脚本是在本地机器上运行，位于 Render 环境之外；然而，在此处，数据库和 Web 服务器都位于相同的 Render 环境中，因此在 main.py 中使用“Internal_Database_Url”。

输入环境变量后，选择“Deploy Web Service”（部署 Web 服务）。

图 15：添加到 Render Web 应用的环境变量

服务部署大约需要几分钟，随后将收到类似下图的通知，显示服务已部署，并提供了一个可访问的 Render 链接。

图 16：Render 上部署的 Web 应用程序

访问此链接将导向“Hello World”方法。如果在此链接末尾添加 /docs，将进入图 17 所示的 Swagger UI 文档界面。在此处，可以使用“Fetch All Users”方法测试并确认 FastAPI Web 应用程序是否已连接到数据库。如下图所示，确实返回了数据。

图 17：Render 上 FastAPI 的 Swagger 文档

最后，需要检查用户推荐系统是否正在动态更新。在之前的 API 调用中，可以看到数据库中存在一个名为“user_username”的用户。使用此用户名调用“Fetch Recommended Game”方法，可以看到最匹配的游戏 appid 为 B08BHHRSPK。

图 18：更新前的用户推荐

通过从游戏列表中随机选择一个 appid = B0BHTKGN7F 的游戏（恰好是“The Elder Scrolls: Skyrim Boardgame”）并利用 user_games 的 POST 方法，可以更新用户的偏好游戏。

图 19：添加一个喜欢的桌面游戏

向用户游戏表添加游戏后，推荐管道理应自动触发，为该用户重新运行并生成新的推荐。如果导航到控制台，可以看到这似乎已经发生，并收到如下所示的“new user recommendations generated”消息。

图 20：推荐管道运行的控制台图像

再次返回 Swagger UI 文档，尝试调用推荐获取方法，可以看到图 21 中显示了与之前不同的推荐列表。这充分表明，随着用户不断添加新数据，推荐管道能够实现自动动态更新，并且其服务已成功在本地环境之外实现无缝访问。

图 21：相似性管道运行后的推荐游戏

总结：

本项目展示了如何搭建和部署一个推荐系统，该系统利用 FastAPI 作为交互层，结合 PostgreSQL 数据库，为用户生成智能的桌面游戏推荐。为了使该推荐系统在未来更加健壮和智能化，可以考虑采取一系列进一步的改进措施，例如在获取更多用户数据后实现混合推荐系统，或者启用用户标签以捕捉更多特征。此外，尽管本文未详细介绍，但项目确实采用了 GitHub workflow，以便在主分支有新更新时自动重建并推送 Docker 镜像，相关代码可在 .github/workflows 中查阅。这极大地加快了开发速度，无需在每次进行小改动时都手动重建 Docker 镜像。

希望本文能为您使用 FastAPI 构建和部署项目提供帮助。

LinkedIn: https://www.linkedin.com/in/lucas-see-6b439188/

Email: seelcs12@gmail.com

图注: 除非另有说明，所有图片均来自作者。

链接: