fastapi使用經驗筆記
venv
python3 -m venv .venv
# 啟動venv
. .venv/bin/activate
# 啟動venv(fish shell)
. .venv/bin/activate.fish
Start App
使用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
在複雜架構的專案中常常需要將一些設定抽出來獨立管理
這時候可以使用官方建議的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
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
可使用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
建立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
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