Flask API 串接 PostgreSQL 實作：用 Postman 測試與資料庫連線完整步驟

Contents 目錄

什麼是API？

API（Application Programming Interface，應用程式介面），是讓不同的應用程式之間可以溝通、交換資料與功能的橋樑

怎麼使用？

假設我今天找到某銀行的公開API，這個 API 每次請求都會回傳「新台幣對外幣」的即時匯率

{
  "base": "TWD",
  "date": "2025-06-07",
  "rates": {
    "USD": 0.033,
    "JPY": 4.761,
    ...
  }
}

我就可以用python request 請求抓取資料，存進資料庫

import requests
import pandas as pd
from datetime import datetime

url = "@@@"
res = requests.get(url)
data = res.json()

# 取出 USD 匯率
usd_rate = data["rates"]["USD"]
today = datetime.today().strftime('%Y-%m-%d')

# 放入 DataFrame
df = pd.DataFrame([{
    "date": today,
    "currency": "USD",
    "rate": usd_rate
}])

from sqlalchemy import create_engine

engine = create_engine("postgresql://user:xxx@localhost:5432/flask_db")
df.to_sql("exchange_rates", engine, if_exists="append", index=False)

也可以用一些套件，每日固定排程抓取，這裡用 Airflow

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def fetch_exchange_rate():
    # 呼叫 API 並寫入資料庫的邏輯放這裡
    pass

dag = DAG("daily_usd_rate", schedule_interval="0 9 * * *", start_date=datetime(2024, 1, 1))

task = PythonOperator(
    task_id="fetch_usd_rate",
    python_callable=fetch_exchange_rate,
    dag=dag
)

資料庫固定有資料後，就可以串接前後端成網站了

什麼是 POST GET PUT/PATCH DELETE？

網站前端（使用者介面）要與網站後端（伺服器資料庫）互動，中間就會使用API來傳遞資料

例如
新增資料，從網站前端頁面送出請求，將新會員資料存進資料庫，這是POST
讀取資料，從網站前端頁面送出請求，查看現有會員資料，這是GET
修改資料，從網站前端頁面送出請求，更新現有會員資料，這是PUT整筆更新/PATCH局部更新
刪除資料，從網站前端頁面送出請求，刪除現有會員資料，這是DELETE

文件撰寫

先設置一個資料夾，照這個順序放，因為我用docker，所以要再加入 Dockerfile、docker-compose.yaml

PS docker 怎麼裝？看前一篇

project_name/
├── app.py
├── requirements.txt
├── Dockerfile
└── docker-compose.yaml

requirements.txt 放所有需要安裝的套件

flask==2.3.2
flask_sqlalchemy==3.1.1
psycopg2-binary==2.9.9

Dockerfile

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 5001
CMD ["python", "app.py"]

docker-compose.yaml
因為跟airflow使用同一個network postgresql，network要指定，ports號也要另外指定，預設8080已經被佔用

# 終端機先cd到之前架airflow的資料夾，查看network，之後指定過去
docker network ls

＃ 這裡出現 airflow_docker_default

version: '3'

services:
  flask-api:
    build: ./flask_api
    ports:
      - "5001:5001"
    networks:
      - airflow_docker_default  # 要跟 airflow 的 network 名稱相同

networks:
  airflow_docker_default:
    external: true

app.py

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

# 連到之前 airflow 設的 PostgreSQL
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://airflow:airflow@postgres:5432/flask_db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    age = db.Column(db.Integer)

@app.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([{"id": u.id, "name": u.name, "age": u.age} for u in users])

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    new_user = User(name=data['name'], age=data['age'])
    db.session.add(new_user)
    db.session.commit()
    return jsonify({"message": "User created"}), 201

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({"error": "User not found"}), 404
    data = request.get_json()
    user.name = data.get('name', user.name)
    user.age = data.get('age', user.age)
    db.session.commit()
    return jsonify({"message": "User updated"})

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({"error": "User not found"}), 404
    db.session.delete(user)
    db.session.commit()
    return jsonify({"message": "User deleted"})

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(host='0.0.0.0', port=5001)

另外還要先在 PostgreSQL 創資料庫

docker-compose exec postgres bash
psql -U airflow

CREATE DATABASE flask_db;
\l  -- 檢查是否建立成功
\q
exit

啟動

＃ 終端機cd到project_name的資料夾
docker-compose restart flask-api

# 重建
docker-compose down
docker-compose up --build

Postman 測試

Postman 官網

載好後 create collection -> create request

POST
照app.py 設的欄位去放置

jsonify([{"id": u.id, "name": u.name, "age": u.age} for u in users])

截圖 2025-06-07 19.02.47

這時再get會看到資料已經被更新了

DELETE

這時再get會看到資料已經被刪除了

用 localhost 查看

http://localhost:5001/users

PS 如果要用 DBeaver 查看

SELECT datname FROM pg_database WHERE datistemplate = false;
SHOW search_path;

SELECT tablename FROM pg_tables WHERE schemaname = 'public';
SELECT * FROM public.users;

PS 看到的數字含義

200 OK : 伺服器成功處理請求（常見於 GET）
201 Created : 成功建立新資源（常見於 POST）
204 No Content : 請求成功，但無需回傳資料（常見於 DELETE）

# 3xx：重導向
301 Moved Permanently : 資源已永久搬移到新位置
302 Found : 暫時搬移，請求會跳轉

# 4xx：客戶端錯誤
400 Bad Request : 請求格式錯誤，可能是缺欄位、JSON 有誤等
401 Unauthorized : 沒有登入授權或 token 無效（通常搭配 JWT）
403 Forbidden : 已登入，但沒有權限使用這個資源
404 Not Found : 指定的資源不存在
409 Conflict : 衝突，例如帳號重複、名稱已存在等
422 Unprocessable Entity : 驗證失敗，例如欄位錯誤或無法處理的輸入（常見於資料驗證）

# 5xx：伺服器錯誤
500 Internal Server Error : 伺服器問題，例如程式錯誤、exception、無法預期錯誤
502 Bad Gateway : 上游伺服器出錯，例如反向代理問題）
503 Service Unavailable : 暫時不能用，通常是伺服器過載或維護中