使用 FastAPI 實作「短網址產生器」應用程式

FastAPI
, ,

前言

本文為 Python 工作坊的基礎教材，以實作「短網址產生器」應用程式為例。

建立專案

建立專案。

1
2
mkdir shortener-python
cd shortener-python

建立虛擬環境

建立虛擬環境。

1
python -m venv .venv

啟動虛擬環境。

1
source .venv/bin/activate

將 .venv 資料夾排除在版本控制之外。

1
echo "*" > .venv/.gitignore

建立 main.py 檔。

1
print('Hello, World!')

初始化版本控制

初始化版本控制。

1
git init

將所有修改添加到暫存區。

1
git add .

提交修改。

1
git commit -m "Initial commit"

指定遠端儲存庫位址。

1
git remote add origin git@github.com:memochou1993/shortener-python.git

推送程式碼到遠端儲存庫。

1
git push -u origin main

安裝 Ruff 格式化工具

新增 requirements.txt 檔。

1
touch requirements.txt

修改 requirements.txt 檔，添加 ruff 依賴套件。

1
ruff

安裝依賴套件。

1
pip install -r requirements.txt

新增 ruff.toml 檔。

1
2
3
4
5
line-length = 120
indent-width = 4

[format]
quote-style = "double"

新增 .vscode/settings.json 檔。

1
2
3
4
5
6
7
8
{
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
        "source.fixAll": "explicit",
        "source.organizeImports": "explicit"
    },
    "editor.defaultFormatter": "charliermarsh.ruff"
}

提交修改。

1
2
git add .
git commit -m "Add ruff dependency"

安裝 FastAPI 框架

修改 requirements.txt 檔，添加 fastapi[standard] 依賴套件。

1
2
# ...
fastapi[standard]

安裝依賴套件。

1
pip install -r requirements.txt

修改 main.py 檔。

1
2
3
4
5
6
7
8
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}

啟動伺服器。

1
fastapi dev main.py

新增 .gitignore 檔，將 __pycache__ 資料夾排除在版本控制之外。

1
__pycache__

提交修改。

1
2
git add .
git commit -m "Add fastapi dependency"

安裝 nanoid 套件

修改 requirements.txt 檔，添加 nanoid 依賴套件。

1
2
# ...
nanoid

安裝依賴套件。

1
pip install -r requirements.txt

修改 main.py 檔。

1
2
3
4
5
6
7
8
9
from fastapi import FastAPI
from nanoid import generate

app = FastAPI()


@app.get("/")
def read_root():
    return {"code": generate(size=8)}

提交修改。

1
2
git add .
git commit -m "Add nanoid dependency"

實作後端

修改 main.py 檔。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from nanoid import generate
from pydantic import BaseModel

app = FastAPI()


links = dict()


class LinkCreateRequest(BaseModel):
    link: str


class LinkCreateResponse(BaseModel):
    code: str


@app.post("/api/links", tags=["Link"], response_model=LinkCreateResponse)
def create_link(body: LinkCreateRequest):
    code = generate(size=8)
    links[code] = body.link

    return LinkCreateResponse(code=code)


@app.get("/{code}", tags=["Link"])
def redirect(code: str):
    link = links.get(code)
    if link is None:
        raise HTTPException(status_code=404, detail="Link not found")

    return RedirectResponse(url=link)

重構

建立 schema/link.py 檔，將請求與回應的結構存放在單獨的檔案。

1
2
3
4
5
6
7
8
9
from pydantic import BaseModel


class LinkCreateRequest(BaseModel):
    link: str


class LinkCreateResponse(BaseModel):
    code: str

建立 schema/__init__.py 檔。

1
（空檔）

修改 main.py 檔，引入請求與回應的結構。

1
2
3
4
5
6
7
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from nanoid import generate

from schema.link import LinkCreateRequest, LinkCreateResponse

# ...

提交修改。

1
2
git add .
git commit -m "Add link creation and redirection functionality"

欄位驗證

修改 schema/link.py 檔。

1
2
3
4
5
6
7
8
9
from pydantic import BaseModel, HttpUrl


class LinkCreateRequest(BaseModel):
    link: HttpUrl


class LinkCreateResponse(BaseModel):
    code: str

提交修改。

1
2
git add .
git commit -m "Add link validation"

實作前端

首先，修改 main.py 檔，試著新增一個會回傳 HTML 內容的端點。

1
2
3
4
5
6
7
8
9
10
11
12
@app.get("/", tags=["Static"], response_class=HTMLResponse)
def read_root():
    return """
    <html>
        <head>
            <title>URL Shortener</title>
        </head>
        <body>
            <h1>Welcome to the URL Shortener</h1>
        </body>
    </html>
    """

重構

新增一個獨立的 static/index.html 檔。

1
2
3
4
5
6
7
8
<html>
  <head>
      <title>URL Shortener</title>
  </head>
  <body>
      <h1>Welcome to the URL Shortener</h1>
  </body>
</html>

修改 main.py 檔，改成讀取 index.html 檔。

1
2
3
4
@app.get("/", tags=["Static"], response_class=HTMLResponse)
def read_root():
    with open(os.path.join("static", "index.html"), "r") as file:
        return HTMLResponse(content=file.read())

實作功能

修改 static/index.html 檔。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>URL Shortener</title>
  <style>
    .container {
      display: flex;
      margin-bottom: 16px;
      max-width: 100%;
      width: 400px;
    }
    #input, #output {
      margin-right: 8px;
      overflow: auto;
      width: 100%;
    }
    button {
      width: 100px;
    }
  </style>
</head>
<body>
  <h1>Welcome to the URL Shortener</h1>
  <div class="container">
    <input id="input" type="text">
    <button id="shorten">Shorten</button>
  </div>
  <div class="container">
    <span id="output"></span>
    <button id="copy" hidden>Copy</button>
  </div>
  <script>
    const input = document.querySelector('#input');
    const output = document.querySelector('#output');
    const shortenButton = document.querySelector('#shorten');
    const copyButton = document.querySelector('#copy');

    shortenButton.addEventListener('click', async () => {
      const link = input.value;
      try {
        const response = await fetch('/api/links', {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json'
          },
          body: JSON.stringify({ link })
        });
        const data = await response.json();
        if (!response.ok) {
          throw new Error(data.detail[0].msg);
        }
        const url = `${window.location.origin}/${data.code}`;
        output.innerHTML = `<a href="${url}" target="_blank" rel="noopener noreferrer">${url}</a>`;
        copyButton.hidden = false;
      } catch (err) {
        console.error(err);
        output.textContent = 'An error occurred. Please try again.';
        copyButton.hidden = true;
      }
    });

    copyButton.addEventListener('click', () => {
      const url = output.querySelector('a').href;
      navigator.clipboard.writeText(url);
    });
  </script>
</body>
</html>

提交修改。

1
2
git add .
git commit -m "Add static page"

實現持久化

首先，到 Supabase 建立一個新專案，並取得 API URL 和 API 金鑰。

建立一個名為 links 的資料表，欄位設置如下：

Column Name Data Type
id int8
created_at timestamptz
code varchar
link text

新增 .env.example 檔。

1
2
SUPABASE_API_URL=
SUPABASE_API_KEY=

新增 .env 檔。

1
2
SUPABASE_API_URL=your-supabase-api-url
SUPABASE_API_KEY=your-supabase-api-key

修改 .gitignore 檔，忽略 .env 檔。

1
2
__pycache__
.env

安裝依賴套件

修改 requirements.txt 檔，添加 supabase 和 python-dotenv 依賴套件。

1
2
3
# ...
supabase
python-dotenv

安裝依賴套件。

1
pip install -r requirements.txt

操作資料庫

修改 main.py 檔，將資料儲存在 Supabase 資料庫中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import os

import supabase
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, status
from fastapi.responses import HTMLResponse, RedirectResponse
from nanoid import generate

from schema.link import LinkCreateRequest, LinkCreateResponse

load_dotenv()


app = FastAPI()


supabase_client = supabase.create_client(
    os.getenv("SUPABASE_API_URL"),
    os.getenv("SUPABASE_API_KEY"),
)


@app.get("/", tags=["Static"], response_class=HTMLResponse)
def read_root():
    with open(os.path.join("static", "index.html"), "r") as file:
        return HTMLResponse(content=file.read())


@app.post("/api/links", tags=["Link"], response_model=LinkCreateResponse)
def create_link(body: LinkCreateRequest):
    code = generate(size=8)

    # 儲存資料到資料庫
    result = (
        supabase_client.from_("links")
        .insert(
            {
                "code": code,
                "link": str(body.link),
            }
        )
        .execute()
    )

    return LinkCreateResponse(**result.data[0])


@app.get("/{code}", tags=["Link"])
def redirect(code: str):
    # 從資料庫取得資料
    result = supabase_client.from_("links").select("*").eq("code", code).execute()

    if not result.data:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thread not found")

    url = result.data[0]["link"]

    return RedirectResponse(url=url)

提交修改。

1
2
git add .
git commit -m "Add Supabase integration"

參考資料