venv

python3 -m venv .venv

# 啟動venv
. .venv/bin/activate

# 啟動venv(fish shell)
. .venv/bin/activate.fish

Start App

ref

使用uvicron啟動web server

透過command line啟動

# 啟動main.py檔案中的app fastapi物件
uvicorn main:app --reload --app-dir src

# 指定app目錄, 例如app entry file在src目錄中
uvicorn main:app --reload --app-dir src

透過python程式啟動

可建立一個start_app.py來跟上述command line進行一樣的啟動方式

好處是如果要增加uvicron的設定不用在command line加一堆很長的指令

import uvicorn

if __name__ == '__main__':
    uvicorn.run(
        app="app:app",
        host='0.0.0.0',
        app_dir="src",
        reload=True
    )

Settings and Environment Variables

ref

在複雜架構的專案中常常需要將一些設定抽出來獨立管理

這時候可以使用官方建議的Pydantic Settings來將這些設定統一做管理

我自己是的習慣建立一個獨立的settings.py

settings.py

這邊可以看到pydantic settings可以對設定檔內的變數做強型態檢查

還可以給預設值

from pydantic import BaseSettings


class Settings(BaseSettings):
    APP_ENV: str = "dev"
    APP_VERSION: str = "0.0.0"
    APP_KEY: str = None

    class Config:
        # 指定讀取.env檔案
        env_file = '.env'

另外也可以搭配python-dotenv做更複雜的設定

動態設定env檔案的來源

from pydantic import BaseSettings
from dotenv import load_dotenv
import os
_mode = os.getenv('MODE', 'dev')
_env_file = os.getenv('ENV_FILE', None)


class Settings(BaseSettings):
    APP_ENV: str = "dev"
    APP_VERSION: str = "0.0.0"
    APP_KEY: str = None
    DOCS: bool = True

    class Config:
        # 在env MODE不為dev的時候讀取/secrets/.env.{_mode}內的檔案(用於cloud run或gke等production環境)
        env_file_path = f'/secrets/.env.{_mode}' if _mode != 'dev' else '.env'
        if _env_file is not None:
            env_file_path = _env_file
        env_file = env_file_path

CORS

refs

fastapi提供CORSMiddleware可以直接做CORS設定

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Swagger Doc設定調整

fastapi預設可直接使用swagger文件

透過FastAPI物件的swagger_ui_parameters選項

即可直接設定swagger configuration

from fastapi import FastAPI
app = FastAPI(
    swagger_ui_parameters={
        "persistAuthorization": True,
        "tryItOutEnabled": True,
    },
)

disable swagger文件

關閉swagger文件只要將FastAPI的docs參數設定為None即可

通常我都是在dev mode啟用swagger在production模式停用

因此我習慣會用settings內的變數來控制

from fastapi import FastAPI
app = FastAPI(
    docs_url="/docs" if settings.DOCS is True else None,
)

Testing

ref

可使用pytest來做測試

fastapi官方則建議可使用httpx套件來對http api來做測試

安裝httpx

pip install httpx

範例

from fastapi.testclient import TestClient
from app import app
client = TestClient(app)
import pytest


@pytest.mark.debug
def test_main_api():
    response = client.get("/")
    assert response.json() == {"Hello": "FastAPI"}

使用pytest-env設定測試時的環境變數及mark

首先安裝pytest-env

pip install pytest-env

接著建立pytest.ini檔案

這樣就可以在執行pytest的時候將環境變數APP_ENV強制設定為testing

並透過設定允許的pytest marker

[pytest]
env =
    APP_ENV=testing

markers =
    debug: debug

執行pytest

pytest

# 顯示print output(預設不顯示)
pytest -s

# 執行單一檔案
pytest src/foobar.py -s

# 執行mark為debug的測試
pytest -k debug -s

Dockerize

FROM python:3.9-slim

ENV APP_SRC /workspace

WORKDIR $APP_SRC
COPY . /$APP_SRC

RUN pip install --no-cache-dir -r requirements.txt

# 執行前面提到的start_app.py
CMD python src/start_app.py

Sqlalchemy連線至MySQL

ref

建立database.py

將db相關設定放在database.py中

from settings import get_setting
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
settings = get_setting()


def get_sqlalchemy_db_url() -> str:
    db_user = settings.DB_USERNAME
    db_pass = settings.DB_PASSWORD
    db_host = settings.DB_HOST
    db_port = settings.DB_PORT
    db_name = settings.DB_NAME
    sqlalchemy_db_url = f"mysql+pymysql://{db_user}:{db_pass}@{db_host}:{db_port}/{db_name}"
    return sqlalchemy_db_url


def get_engine():
    engine = create_engine(
        get_sqlalchemy_db_url(), connect_args={}
    )
    return engine


def get_db_session():
    db = sessionmaker(autocommit=False, autoflush=False, bind=get_engine())()
    try:
        yield db
    finally:
        db.close()

執行Query

from database import get_db_session
db = next(get_db_session())
db.execute("SELECT * FROM foobar")

透過unix socket連線至CloudSQL

ref

sqlalchemy_db_url使用以下格式即可透過unix socket連線至cloud sql

CLOUD_SQL_SOCKET_PATH格式為"/cloudsql/{PROJECT_ID}:{REGION}:{INSTANCE_NAME}"

例如: /cloudsql/project-foobar:asia-east1:my-dev-db

sqlalchemy_db_url = f"mysql+pymysql://{db_user}:{db_pass}@?unix_socket={CLOUD_SQL_SOCKET_PATH}"

設定session的isolation level

透過create_engine的isolation_level參數即可設定

engine = create_engine(
    get_sqlalchemy_db_url(), connect_args={}, isolation_level="READ COMMITTED"
)

alembic

安裝

pip install alembic

初始化

後面的alembic是指輸出的目錄

因此執行此指令後會建立一個alembic目錄

裡面包含一些alembic的設定

並且會在執行的目錄下建立一個alembic.ini檔案

alembic init alembic

建立一個載入所有model的python檔案

通常我會建立一個models目錄

並在底下建立一個__init__.py內容大致如下

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import MetaData

meta = MetaData()
Base = declarative_base(metadata=meta)

# 以下載入所有models

調整env.py

找到"fileConfig(config.config_file_name)"在它下方加入以下這段

將覆蓋掉原本的sqlalchemy url

config.set_main_option("sqlalchemy.url", url)

載入models/init.py

並將target_metadata改成Base.metadata

from models import *
target_metadata = Base.metadata

自動產生migration file

使用以下指令即可自動產生migration檔案在alembic/versions目錄下

有需要的話可以再自行調整migration格式

alembic revision --autogenerate -m "message"

其他常用alembic指令

# migrate
alembic upgrade head

# rollback
alembic downgrade -1

# 執行時設定env
APP_ENV=testing alembic upgrade head

fastapi使用經驗筆記