Python / backend / fastapi === ###### tags: `Python` ###### tags: `Python`, `PyPi`, `backend`, `fastapi` <br> [TOC] <br> ## intro ### 優勢 - [How Python's WSGI vs. ASGI is Like Baking a Cake](https://developer.vonage.com/en/blog/how-wsgi-vs-asgi-is-like-baking-a-cake) ![](https://hackmd.io/_uploads/SJi05Qa_p.png) ### Tools - [Uvicorn](https://www.uvicorn.org/#config-and-server-instances) - [[github] s3rius/FastAPI-template](https://github.com/s3rius/FastAPI-template) - [[github] andfanilo/fastapi-vue-crud](https://github.com/andfanilo/fastapi-vue-crud) ### 教學資料 - [fastapi使用經驗筆記](https://ciao-chung.com/page/article/fastapi-note) - [FastAPI（2）快速入门](https://blog.csdn.net/ben_na_/article/details/124003462) - [FastAPI - www.slingacademy.com](https://www.slingacademy.com/cat/fastapi/#google_vignette) - [FastAPI基础：Depends怎么用？](https://blog.csdn.net/NeverLate_gogogo/article/details/112472480) :+1: <br> ## API keys - ### [A Beginner’s Guide to Using API Keys in FastAPI and Python](https://medium.com/@valerio.uberti23/256fe284818d) - ### [[starlette] Authentication](https://www.starlette.io/authentication/) <br> ## Migration Guide - [Required, optional, and nullable fields](https://docs.pydantic.dev/latest/migration/#required-optional-and-nullable-fields) <br> ## demo ### demo1 - simple - ### 安裝 ``` $ pip install fastapi $ pip install uvicorn ``` - `app.py` ```python= from fastapi import FastAPI app = FastAPI() @app.get('/') def hello(): return 'hello' ``` - ### terminal ``` $ uvicorn app:app --host 0.0.0.0 --port 38888 ``` [![](https://i.imgur.com/3UnkFLB.png)](https://i.imgur.com/3UnkFLB.png) - ### browser ![](https://i.imgur.com/fxSUezQ.png) ### demo2 - reload + APIRouter - reload reload 功能就是可以一邊改 code，一邊看效果，不用重啟 app - `main.py` ```python= import uvicorn from fastapi import FastAPI, APIRouter, Request app = FastAPI() router = APIRouter() @router.get('/') def index(request: Request): return 'home' @router.get('/hello') def hello(request: Request): return 'hello' # include_router must be placed after @router app.include_router(router) # or prefix='' if __name__ == '__main__': uvicorn.run( "main:app", host="localhost", port=8000, reload=True, ) ``` - 不能將 `main.py` 命名為 `__main__.py`，然後用 `"__main__:app"` 啟動，這樣會有 error ``` ERROR: Error loading ASGI app. Attribute "app" not found in module "__main__". ``` - 啟用 reload 功能 `uvicorn.run("main:app", ...)` 中的 "main:app" 必須是字串 不能是 `uvicorn.run(app, ...)` (直接引用 app) 否則會有底下 error: ``` WARNING: You must pass the application as an import string to enable 'reload' or 'workers'. 警告：您必須以匯入字串的形式傳遞應用程式，才能啟用「重載」或「工作者」功能。 ``` <br> <br> ## app lifecycle ### [Lifespan Events](https://fastapi.tiangolo.com/advanced/events/#startup-event) - [startup event](https://fastapi.tiangolo.com/advanced/events/#startup-event) ```python= @app.on_event("startup") async def startup_event(): items["foo"] = {"name": "Fighters"} items["bar"] = {"name": "Tenders"} ``` <br> ## read request-body - [How to read body as any valid json?](https://stackoverflow.com/questions/64379089/how-to-read-body-as-any-valid-json) ```python= from fastapi import FastAPI, Request, Response import uvicorn app = FastAPI() @app.post('/post') async def post(request: Request): print('request:', request) print('type(request):', type(request)) return await request.json() if __name__ == '__main__': uvicorn.run(app, port=8081) ``` - 執行方式 - server-side `$ python main.py` - client-side `$ curl -X POST 127.0.0.1:8081/post -d '{"msg": "hello"}'` - 如果沒有 await/async 會有底下錯誤 ``` INFO: 127.0.0.1:55936 - "POST /post HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application Traceback (most recent call last): File ".../site-packages/fastapi/encoders.py", line 322, in jsonable_encoder data = dict(obj) TypeError: 'coroutine' object is not iterable During handling of the above exception, another exception occurred: Traceback (most recent call last): File ".../site-packages/fastapi/encoders.py", line 327, in jsonable_encoder data = vars(obj) TypeError: vars() argument must have __dict__ attribute The above exception was the direct cause of the following exception: Traceback (most recent call last): File ".../site-packages/uvicorn/protocols/http/h11_impl.py", line 408, in run_asgi result = await app( # type: ignore[func-returns-value] File ".../site-packages/uvicorn/middleware/proxy_headers.py", line 84, in __call__ return await self.app(scope, receive, send) File ".../site-packages/fastapi/applications.py", line 1106, in __call__ await super().__call__(scope, receive, send) File ".../site-packages/starlette/applications.py", line 122, in __call__ await self.middleware_stack(scope, receive, send) File ".../site-packages/starlette/middleware/errors.py", line 184, in __call__ raise exc File ".../site-packages/starlette/middleware/errors.py", line 162, in __call__ await self.app(scope, receive, _send) File ".../site-packages/starlette/middleware/exceptions.py", line 79, in __call__ raise exc File ".../site-packages/starlette/middleware/exceptions.py", line 68, in __call__ await self.app(scope, receive, sender) File ".../site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__ raise e File ".../site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__ await self.app(scope, receive, send) File ".../site-packages/starlette/routing.py", line 718, in __call__ await route.handle(scope, receive, send) File ".../site-packages/starlette/routing.py", line 276, in handle await self.app(scope, receive, send) File "INFO: 127.0.0.1:55936 - "POST /post HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application Traceback (most recent call last): File ".../site-packages/fastapi/encoders.py", line 322, in jsonable_encoder data = dict(obj) TypeError: 'coroutine' object is not iterable During handling of the above exception, another exception occurred: Traceback (most recent call last): File ".../site-packages/fastapi/encoders.py", line 327, in jsonable_encoder data = vars(obj) TypeError: vars() argument must have __dict__ attribute The above exception was the direct cause of the following exception: Traceback (most recent call last): File ".../site-packages/uvicorn/protocols/http/h11_impl.py", line 408, in run_asgi result = await app( # type: ignore[func-returns-value] File ".../site-packages/uvicorn/middleware/proxy_headers.py", line 84, in __call__ return await self.app(scope, receive, send) File ".../site-packages/fastapi/applications.py", line 1106, in __call__ await super().__call__(scope, receive, send) File ".../site-packages/starlette/applications.py", line 122, in __call__ await self.middleware_stack(scope, receive, send) File ".../site-packages/starlette/middleware/errors.py", line 184, in __call__ raise exc File ".../site-packages/starlette/middleware/errors.py", line 162, in __call__ await self.app(scope, receive, _send) File ".../site-packages/starlette/middleware/exceptions.py", line 79, in __call__ raise exc File ".../site-packages/starlette/middleware/exceptions.py", line 68, in __call__ await self.app(scope, receive, sender) File ".../site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__ raise e File ".../site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__ await self.app(scope, receive, send) File ".../site-packages/starlette/routing.py", line 718, in __call__ await route.handle(scope, receive, send) File ".../site-packages/starlette/routing.py", line 276, in handle await self.app(scope, receive, send) File ".../site-packages/starlette/routing.py", line 66, in app response = await func(request) File ".../site-packages/fastapi/routing.py", line 292, in app content = await serialize_response( File ".../site-packages/fastapi/routing.py", line 180, in serialize_response return jsonable_encoder(response_content) File ".../site-packages/fastapi/encoders.py", line 330, in jsonable_encoder raise ValueError(errors) from e ValueError: [TypeError("'coroutine' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')] /site-packages/starlette/routing.py", line 66, in app response = await func(request) File ".../site-packages/fastapi/routing.py", line 292, in app content = await serialize_response( File ".../site-packages/fastapi/routing.py", line 180, in serialize_response return jsonable_encoder(response_content) File ".../site-packages/fastapi/encoders.py", line 330, in jsonable_encoder raise ValueError(errors) from e ValueError: [TypeError("'coroutine' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')] ``` <br> <br> <hr> <br> ## 下載檔案 ### 提供暫時的下載 link - ### [ChatGPT] 需求：後端設計，請寫一個簡單範例，能提供給使用者暫時link, 讓使用者可以下載檔案，一段時間後該 link 會失效，並請避免 concurrency 問題<br>限制：請使用 fastapi 套件 下面是一個使用FastAPI套件的簡單範例，可以提供使用者一個暫時的下載連結，該連結在一段時間後會失效，並且避免了並發問題： ```python= from fastapi import FastAPI from starlette.responses import FileResponse from datetime import datetime, timedelta import uuid app = FastAPI() # 設定連結的有效期限（以秒為單位） LINK_EXPIRATION_TIME = 3600 # 1小時 # 儲存暫時連結的字典 temporary_links = {} @app.get('/download/{link_id}') async def download_file(link_id: str): # 檢查連結是否有效 if link_id in temporary_links: current_time = datetime.now() expiration_time = temporary_links[link_id] # 檢查連結是否過期 if current_time < expiration_time: # 暫時連結有效，提供下載檔案 file_path = '/path/to/file.txt' # 替換為實際檔案路徑 return FileResponse(file_path, media_type='application/octet-stream', filename='file.txt') # 連結無效或已過期，回傳錯誤訊息 return {"message": "連結無效或已過期"} @app.get('/generate_link') async def generate_download_link(): # 生成一個隨機的連結ID link_id = str(uuid.uuid4()) # 儲存當前時間與過期時間 current_time = datetime.now() expiration_time = current_time + timedelta(seconds=LINK_EXPIRATION_TIME) # 儲存暫時連結的過期時間 temporary_links[link_id] = expiration_time # 回傳包含連結ID的下載連結 download_link = f"/download/{link_id}" return {"download_link": download_link} if __name__ == '__main__': import uvicorn uvicorn.run(app) ``` 在這個範例中，使用者可以透過訪問`/generate_link`路由來獲取一個暫時的下載連結。該連結將包含一個隨機生成的連結ID。當使用者訪問下載連結`/download/{link_id}`時，會檢查連結的有效性和過期時間。如果連結有效，將提供下載檔案。否則，將返回錯誤訊息。 範例中使用了`temporary_links`字典來儲存暫時連結的過期時間。每次生成連結時，會生成一個連結ID並與過期時間一起儲存。當使用者訪問下載連結時，會檢查連結的有效 - ### [ChatGPT] 上面的程式碼，如果同時間有2個request一起進來，會發生啥事? 在上面的程式碼中，如果同時間有兩個請求同時進來，可能會發生以下情況： - 競爭條件（Race Condition）：由於Python的GIL（Global Interpreter Lock）限制，同一時間只有一個線程可以執行Python的原生碼。這意味著當多個請求同時進來時，它們將按照請求的順序被處理。因此，在同一時間內，只會有一個請求進行處理，而另一個請求將等待。 - 共享狀態：如果多個請求同時存取和修改temporary_links字典，可能會導致共享狀態的問題。例如，當兩個請求同時執行generate_download_link()時，它們可能會同時生成相同的連結ID，並將相同的連結ID存儲到temporary_links字典中。這可能導致其中一個請求的連結失效，因為它被另一個請求覆蓋。 為了解決這些問題，你可以考慮使用同步機制，例如鎖（Lock）或信號量（Semaphore），以確保在修改共享資源時只有一個請求可以訪問。此外，你也可以考慮使用更強大的資料庫或緩存系統來管理連結的生成和過期時間。這些方法可以提供更可靠和穩定的暫時連結功能。 - ### [ChatGPT] 請引入同步機制( lock 或 semaphore 機制或其他用法) 到上述程式碼的共享資源 為了引入同步機制到上述程式碼的共享資源，你可以使用threading模組中的Lock或Semaphore來實現。以下是一個修改後的程式碼範例，使用Lock同步機制： ```python= import time from fastapi import FastAPI from threading import Lock app = FastAPI() lock = Lock() temporary_links = {} @app.get("/download/{file_id}") def download_file(file_id: str): if file_id in temporary_links: return {"message": "Downloading file with ID: {}".format(file_id)} else: return {"message": "File with ID {} not found".format(file_id)} @app.post("/generate_link/{file_id}") def generate_download_link(file_id: str, expiration_time: int): with lock: if file_id not in temporary_links: temporary_links[file_id] = expiration_time return {"message": "Generated download link for file ID: {}".format(file_id)} else: return {"message": "Download link for file ID {} already exists".format(file_id)} def check_link_expiration(): while True: with lock: expired_files = [file_id for file_id, expiration_time in temporary_links.items() if expiration_time <= time.time()] for file_id in expired_files: del temporary_links[file_id] time.sleep(60) # Check link expiration every 60 seconds if __name__ == "__main__": import threading expiration_thread = threading.Thread(target=check_link_expiration) expiration_thread.start() ``` 上述程式碼在generate_download_link()和check_link_expiration()函式中使用with lock:語句來確保在修改temporary_links字典時只有一個線程可以訪問。這樣可以避免競爭條件和共享資源的問題。 需要注意的是，Lock是一種簡單的同步機制，但在高併發情況下可能會導致效能下降。如果需要更高效的同步機制，可以考慮使用Semaphore或其他進階的同步機制，具體取決於你的需求和使用情境。 <br> ## FastAPI vs Flask ### flask 和 fastAPI 的 SWAT 分析 [ChatGPT] SWAT 分析通常是指 Strengths（優勢）、Weaknesses（劣勢）、Opportunities（機會）、Threats（威脅）的分析，是一種在制定策略或評估特定方案時用到的分析工具。然而，在你的提問中，可能有一些誤解，因為 SWAT 不是常見於比較 Flask 和 FastAPI 這類框架的評估指標。 如果你想要比較 Flask 和 FastAPI，可以從以下角度進行評估： - **Flask:** - **優勢：** - 簡單易學： Flask 是一個輕量級框架，易於入門，適合初學者。 - 靈活性： 提供了基本的工具，但也允許開發者使用其他庫和工具，以滿足特定需求。 - 劣勢：** - 異步性能： 在處理大量並發請求時，可能缺乏某些異步性能。 - **FastAPI:** - **優勢：** - 異步支持： 提供了基於 Starlette 的異步支持，有助於處理大量並發請求。 - 型別標註： 使用 Pydantic 庫進行輸入和輸出的型別標註，提供更好的自文檔和自動驗證。 - **劣勢：** - 較新： 相較於 Flask，FastAPI 是一個相對較新的框架，可能在某些方面缺少社群和生態系統的支持。 - **共同特點：** - RESTful： 兩者都支持 RESTful 風格的 API 設計。 - Python： 都是使用 Python 語言。 - **選擇標準：** - 專案需求： 要根據具體專案需求來選擇框架。如果是較簡單的應用，可能 Flask 就足夠。對於需要處理大量並發和更豐富功能的應用，FastAPI 可能更適合。 以上僅為簡單的比較，實際選擇應視具體情況而定。 <br> ## 使用代理 ### [教學] http://www.itdocs.icu/fastapi/ - [使用代理](http://www.itdocs.icu/fastapi/advanced/behind-a-proxy/) <br> ## Troubleshooting - [Adding async to Fastapi function gives me the "'coroutine' object is not iterable" error](https://stackoverflow.com/questions/72611531/) async function + await call 相關 error log: ``` INFO: 127.0.0.1:38594 - "POST /text-generation/api/models/conversation HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application Traceback (most recent call last): File "/home/tj/miniconda3/envs/python3.9/lib/python3.9/site-packages/fastapi/encoders.py", line 322, in jsonable_encoder data = dict(obj) TypeError: 'coroutine' object is not iterable During handling of the above exception, another exception occurred: Traceback (most recent call last): File ".../python3.9/site-packages/fastapi/encoders.py", line 327, in jsonable_encoder data = vars(obj) TypeError: vars() argument must have __dict__ attribute The above exception was the direct cause of the following exception: Traceback (most recent call last): File ".../python3.9/site-packages/uvicorn/protocols/http/h11_impl.py", line 408, in run_asgi result = await app( # type: ignore[func-returns-value] ... File ".../python3.9/site-packages/fastapi/encoders.py", line 330, in jsonable_encoder raise ValueError(errors) from e ValueError: [TypeError("'coroutine' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')] ``` - 修改變化 - 修改前： ``` async def conversation(request: Request): return send_request(request, chat=True) # <- 少了 await ``` - 修改後： ``` async def conversation(request: Request): return await send_request(request, chat=True) ``` <br> ## 參考資料 - [[day8]FASTAPI建置示範-進度落後了!!](https://ithelp.ithome.com.tw/articles/10270523) - FastAPI 是一個容易上手，並且可以快速開發高效能 WEB API 的框架 - uvicorn是一個輕量級高效的web伺服器框架 - [请不要把 Flask 和 FastAPI 放到一起比较](https://zhuanlan.zhihu.com/p/369591096) - [2023/6/8 8:21] FastAPI + SQLAlchemy - https://fastapi.tiangolo.com/tutorial/sql-databases/ - https://hackmd.io/@YMont/python-fastapi-2 - https://blog.devgenius.io/fast-api-with-db-connection-pools-cdfd6000827 - https://patrick-muehlbauer.com/articles/fastapi-with-sqlalchemy