在 Fly.io 上使用 SQLite 部署 FastAPI 應用程式

雲端解決方案適用於中型和大型項目，但對於小型個人專案來說太重了。如果你想啟動一些小東西（一些 api 端點和一個小儲存庫），有三個選項：

• 使用與「大型」專案（AWS ECS/EKS、RDS）相同的方法，但它們是多餘的，並且基礎設施代碼可能比實際專案的程式碼更大。而且價格也很貴（~$100）。
• 使用無伺服器解決方案（Lambda、Vercel）。大多數雲端供應商都有這樣的解決方案，但這些服務在簡單資料庫方面存在困難- 他們提供廉價的供應商解決方案（AWS）或需要託管資料庫，這又是昂貴的（對於無伺服器來說幾乎沒有什麼，資料庫約20 美元）
• 將 VPS 與 Docker 結合使用。它很便宜（小型機器約 5 美元），幾乎不需要管理基礎設施，但部署很糟糕（需要私人或自託管註冊表、來自 CI 的 SSH 訪問）。

我通常使用 SQLite 編寫小型應用程序，它是一個方便的小型單文件資料庫，可以使用任何程式語言運行，並且可以複製到本地電腦以分析資料等。因此，我一直在尋找一些結合了無服務方法、易於部署和使用 SQLite 的能力的中間件解決方案，並找到了 Fly.io。

設定

如果您在 Fly.io 中沒有帳戶 – 您需要建立帳戶。管理專案還需要名為 Flyctl 的 CLI 工具。 Fly.io 既可以在本地部署，也可以從 CI 部署。

flyctl 透過 Dockerfile 從專案的根資料夾部署，這很酷，因為相同的 Dockerfile 可以在其他系統中使用。為了玩 Fly.io，我準備了一個簡單的 FastAPI 項目，將狀態儲存在資料庫中 - 具有點擊計數功能的通用 url 縮短器。

Dockerfile：

FROM python:3.13-alpine
WORKDIR /app

COPY ./requirements.txt .
RUN pip install --no-cache-dir --upgrade -r requirements.txt

COPY . /app

ENV HOST=0.0.0.0 PORT=8080
EXPOSE ${PORT}
CMD uvicorn main:app --host ${HOST} --port ${PORT}

main.py:

import asyncio
import random
import string
from urllib.parse import urlparse

import aiosqlite
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse

DB_PATH = "/data/app.db"

app = FastAPI()

async def get_db() -> aiosqlite.Connection:
    if db := getattr(get_db, "_db", None):
        if db.is_alive:
            return db

    db = await aiosqlite.connect(DB_PATH, loop=asyncio.get_event_loop())
    db.row_factory = aiosqlite.Row

    qs = """
    CREATE TABLE IF NOT EXISTS links (
        created_at INTEGER DEFAULT (strftime('%s', 'now')),
        short_code TEXT PRIMARY KEY,
        full_url TEXT NOT NULL,
        clicks INTEGER DEFAULT 0
    )
    """

    await db.execute(qs)
    await db.commit()

    setattr(get_db, "_db", db)
    return db

def random_code(length=8) -> str:
    alphabet = string.ascii_letters + string.digits
    return "".join(random.choice(alphabet) for x in range(length))

def is_valid_url(url: str) -> bool:
    try:
        parts = urlparse(url)
        return all([parts.scheme, parts.netloc])
    except ValueError:
        return False

@app.post("/")
async def shorten(url: str, req: Request):
    if not is_valid_url(url):
        raise HTTPException(status_code=400, detail="Invalid URL")

    host = req.headers.get("host")
    if host is None:
        raise HTTPException(status_code=500, detail="Missing host header")

    short_code = random_code()
    db = await get_db()
    qs = "INSERT INTO links (short_code, full_url) VALUES (?, ?)"
    await db.execute(qs, (short_code, url))
    await db.commit()

    return f"https://{host}/{short_code}"

@app.get("/")
async def list_links():
    db = await get_db()
    qs = "SELECT short_code, full_url, clicks FROM links ORDER BY created_at DESC"
    async with db.execute(qs) as cursor:
        return await cursor.fetchall()

@app.get("/{short_code}")
async def redirect(short_code: str):
    db = await get_db()
    qs = """
    UPDATE links SET clicks = clicks + 1 WHERE short_code = ?
    RETURNING full_url
    """

    async with db.execute(qs, (short_code,)) as cursor:
        if row := await cursor.fetchone():
            return RedirectResponse(row["full_url"])

    raise HTTPException(status_code=404)

需求.txt：

aiosqlite
fastapi
uvicorn

部署

要部署我們的程式碼，首先我們需要建立一個 Fly.io 專案。這可以在 Web 介面或使用 Flyctl 中完成。若要使用 CLU 工具在根資料夾（程式碼所在的位置）中建立項目，應執行 Flyctl launch。此命令將提供選擇所需的硬體並建立 Fly.toml 檔案：

fly launch --build-only

您將來可以透過更改此文件中的參數或透過 Web ui 來修改項目。基本的fly.toml看起來不錯，但SQLite需要存儲，可以使用以下命令創建：

fly volumes create sqlite_data -s 1 -r ams

其中 -s 1 將磁碟區大小設為 1 GB（預設為 3 GB），-r 是將建立磁碟區的區域（使用與建立 Fly.io 專案相同的區域）。您以後可以隨時變更儲存大小。

最後要做的事情是向 Fly.toml 添加一個 mounts 部分，它將卷附加到應用程式：

FROM python:3.13-alpine
WORKDIR /app

COPY ./requirements.txt .
RUN pip install --no-cache-dir --upgrade -r requirements.txt

COPY . /app

ENV HOST=0.0.0.0 PORT=8080
EXPOSE ${PORT}
CMD uvicorn main:app --host ${HOST} --port ${PORT}

sqlite_data 是儲存的名稱，/data 是連接磁碟區的路徑。這本質上與 docker run --mount source=sqlite_data,target=/data 或對應的 Docker Compose 部分相同。

SQLite 無法從多個應用程式寫入，而 Fly.io 預設會為一個應用程式建立 2 個實例，因此我們可以將副本數量指定為 1，以防萬一：

import asyncio
import random
import string
from urllib.parse import urlparse

import aiosqlite
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse

DB_PATH = "/data/app.db"

app = FastAPI()

async def get_db() -> aiosqlite.Connection:
    if db := getattr(get_db, "_db", None):
        if db.is_alive:
            return db

    db = await aiosqlite.connect(DB_PATH, loop=asyncio.get_event_loop())
    db.row_factory = aiosqlite.Row

    qs = """
    CREATE TABLE IF NOT EXISTS links (
        created_at INTEGER DEFAULT (strftime('%s', 'now')),
        short_code TEXT PRIMARY KEY,
        full_url TEXT NOT NULL,
        clicks INTEGER DEFAULT 0
    )
    """

    await db.execute(qs)
    await db.commit()

    setattr(get_db, "_db", db)
    return db

def random_code(length=8) -> str:
    alphabet = string.ascii_letters + string.digits
    return "".join(random.choice(alphabet) for x in range(length))

def is_valid_url(url: str) -> bool:
    try:
        parts = urlparse(url)
        return all([parts.scheme, parts.netloc])
    except ValueError:
        return False

@app.post("/")
async def shorten(url: str, req: Request):
    if not is_valid_url(url):
        raise HTTPException(status_code=400, detail="Invalid URL")

    host = req.headers.get("host")
    if host is None:
        raise HTTPException(status_code=500, detail="Missing host header")

    short_code = random_code()
    db = await get_db()
    qs = "INSERT INTO links (short_code, full_url) VALUES (?, ?)"
    await db.execute(qs, (short_code, url))
    await db.commit()

    return f"https://{host}/{short_code}"

@app.get("/")
async def list_links():
    db = await get_db()
    qs = "SELECT short_code, full_url, clicks FROM links ORDER BY created_at DESC"
    async with db.execute(qs) as cursor:
        return await cursor.fetchall()

@app.get("/{short_code}")
async def redirect(short_code: str):
    db = await get_db()
    qs = """
    UPDATE links SET clicks = clicks + 1 WHERE short_code = ?
    RETURNING full_url
    """

    async with db.execute(qs, (short_code,)) as cursor:
        if row := await cursor.fetchone():
            return RedirectResponse(row["full_url"])

    raise HTTPException(status_code=404)

所有設定現已完成，我們可以使用命令部署我們的應用程式：

aiosqlite
fastapi
uvicorn

應用程式應成功啟動，公共 DNS 名稱將列印到控制台。現在我們可以透過發布一些縮短器的網址來檢查它：

fly launch --build-only

然後我們可以訪問這個鏈接，它應該重定向到 https://example.com。最後，我們可以檢查點擊次數是否已更新：

fly volumes create sqlite_data -s 1 -r ams

要檢查部署之間保存的資料庫狀態，我們可以使用 FlyDeploy 執行新部署，並​​檢查鏈接列表是否與上面相同（1 個鏈接，1 個點擊）。

遷移

如果您使用外部解決方案進行遷移，而不是在應用程式啟動時從程式碼執行它們，那麼運行遷移的唯一方法是將其作為 RUN 命令的一部分放入 Dockerfile 中。

備份

我們可以使用 Fly ssh 控制台連接到機器，然後在 /data 資料夾中與資料庫檔案互動。我們也可以使用以下命令將資料庫檔案複製到本機：

[mounts]
source = "sqlite_data"
destination = "/data"

結論

Fly.io 是一項簡單且方便的應用程式部署服務。從 Docker 容器進行部署，附加服務包括 PSQL、Redis、S3 等儲存（與 Vercel 不同）。它很便宜，最便宜的服務費用為3 美元（1 個共享CPU / 256 MB） - 如果流量很少，容器可能會更少- 容器在沒有活動的幾分鐘後關閉，並在出現流量時自動打開。

缺點是，沒有針對計劃任務的內建解決方案 - 相反，官方解決方案是使用 crontab 設定單獨的伺服器並從中運行任務 - 這有點令人毛骨悚然。

以上是在 Fly.io 上使用 SQLite 部署 FastAPI 應用程式的詳細內容。更多資訊請關注PHP中文網其他相關文章！