【GCP】Pub/Sub Python 實戰紀錄 | DevFest 2023 台中工作坊

前言
此篇文章為 2023/12/09 DevFest Taichung Serverless workshop 步驟文章,如果有需要透過 GCP Pub/Sub 將訊息轉打給訂閱的 Cloud Run endpoint,可以參考看看這篇文章喔!
標籤

此篇文章為 2023/12/09 DevFest Taichung Serverless workshop 步驟文章,如果有需要透過 GCP Pub/Sub 將訊息轉打給訂閱的 Cloud Run endpoint,可以參考看看這篇文章喔!

近期總算有空寫一些 side project,當然同時也會來 maintain 有在使用的專案,但因為過去都使用 Flask(近期都 FastAPI),有些Code年久失修,或是有升級版本,都會造成 Cloud Run 上的佈署失敗,以下就提供解法給大家。

本次範例專案: https://github.com/louis70109/Python-import-env-json-string

這一陣子許多同事除了在內部分享會中分享 Testcontainer 的技術,也有到 JCConf 2021 上分享 - Integration testing with Testcontainers,過往我在寫單元測試(Unit)時,時常都把資料庫的函式都直接擋住(Mock),假裝他是成功的情況下往下走,但在一個 API 中資料庫往往是最容易在緊繃時掛掉,亦或是工程是手癢去改了某個欄位,讓整個服務瞬間炸掉,這些問題都是我們無法控制的部分...因此實際讓資料庫跑在測試時可以有個真的環境可以使用是很重要的一環,雖然看專案規模,測試時間有多有少有長有短,但放著一個 Container/VM 在那邊沒用就是很浪費啊,因此就有本次要介紹的 Test Container。
把 Test Container 分開寫比較懂他是幹嘛的,但在找資料時請以 testcontainer 去找才找得到喔!
Test Container 是一個 Java 的函式庫,支援 JUnit 來做測試,在跑測試時可以幫忙起個容器(Container),並可以放上各式資料庫、Selenium 瀏覽器...等等可以放在 Docker 上面跑的內容,那 Test Container 很好的地方就是他也提供 Python 的解決方案(testcontainers/testcontainers-python),讓我這個 Python 開發者可以有機會在 Docker 裡面很快地起一個 Container 來跑資料庫相關的整合測試,接下來就讓我娓娓道來我踩坑的過程 🖌️
有支援各種程式語言的函示庫喔,大家可以去挖挖看
故事是這樣的,前幾天好像有微軟的開發者日(❓),因此同事在收看完線上影片後分享給我一篇關於在 VSCode 開發 FastAPI+PostgreSQL 的影片,內容主要在講解如何在 VSCode 中以 Container 模式做開發,詳細內容以下會提到。
而因為過往我因為很懶得處理環境問題,因此都快速安裝 Python 後在使用 PyCharm 來協助處理開發問題,在這次看了影片中的使用覺得整體效果非常讚(香 🥰),以下就來介紹一下如何使用以及範例吧!
官方範例是用 Poetry 來管理套件,我則習慣使用 PIP + requirements.txt 來管理
延伸閱讀:

近期收到了一個來自 Lotify 的 Pull Request,主要功能是加入 Command Line 的功能,讓用戶在安裝 Lotify 之後即可在終端機上測試,感謝來自熱情的網友幫忙。
原以為 Merge 之後應該要可以用了,於似乎開始安裝要開始使用測試,沒想到遇到了一個錯誤
ModuleNotFoundError: No module named 'lotifyCli'
於是就開始的抓蟲之旅。✍️
從我開始工作時,因為動態語言方便且快速上手,因此我就選擇動態語言作為我吃飯的工具。隨著時間的推進我寫過了 Python、Ruby、JavaScript,由於大多都使用 VScode 撰寫程式,對於環境設定不是很熟捻的我設定相關東西就很困擾,因此大多數都是用 print(Python)、console.log(JS) 或是 puts(Ruby) 直接印在終端機上來除錯腳本或是網站。
在過去這些日子中最有印象的就是使用 Ruby 的 byebug 套件來幫我除蟲,它是個可以在終端機透過快捷鍵讓你取得想除錯的地方開始進行除錯,對於當時的我來說簡直是福音,也讓我在那段日子中非常的快樂XD
I wrote dynamic programming language from working to now. Most of my time also use print(Python), console.log(JS), puts(Ruby) to debug any scripts or web apps. from my before experience, I only know Ruby had byebug package which could write a byebug line in your code and run it, it will show something information that you could debug in the terminal.
This is my first know debug method in dynamic languages ... XD
雖然現在網路越來越快,每個同步(Synchronize)的 request 處理速度都很快,但用非同步(Asynchronous)不僅可以同時處理較多 request,也不會被前面那個 request 拖時間導致後面排隊的 request 不用做事,各走各的路,出事自己負責(咦?),雖然非同步有其他也要探討的問題,但這裡就先不討論這個~

過往在寫應用程式時總是為了趕上線 將程式寫完之後即部署,但往往總是上線後才發生了一些可預期的錯誤,導致工時增加,因此在撰寫程式時若有自動測試的程式碼,透過測試增加來程式碼的品質,為上線前做一個最後的把關。
而近期在寫 Side Project - P+ League Bot 時包裝了一些程式碼(FlexMessage、SQL Query...),但再將資料庫函式改成 ORM - SQLAlchemy 時導致了一些上線錯誤,因此就開始補測試程式碼避免往後再有此類問題產生,本篇則紀錄使用 Pytest 時的一些小知識。
@startuml
專案->Pytest: 本地端執行測試
Pytest->專案: 本地測試完成(成功/失敗)
專案->GitHub: 送上倉庫
GitHub->GitHubActions: 觸發 CI
GitHubActions->GitHub: 線上 CI 測試完成(成功/失敗)
GitHub->Heroku: 測試成功即部署
@enduml

在過往很多開源專案大家都會使用 Travis (也真的好用),但隨著 GitHub Actions 推出之後也許多人轉移上去,畢竟讓資源都在同一個平台上也比較好管理。而最近看到許多專案(LINE SDK)都開始改到 Actions 上,且因緣際會下看到 GitHub 的文件上敘述如何使用,而在個人專案上也能很清楚明瞭的點選到相關的 Actions flow,所以本次就用這篇來介紹一下啦~
範例專案請參考我的 GitHub

從一開始在社群中為了 demo code 將程式放上 GitHub,接著到了認識前輩們都在開源圈積極展示自己的成果也分享帶來的好處,造就了我也加入開源行列,但看了 open source 中許多好用的工具,也希望自己能夠看懂並加入開發,希望自己的貢獻能力(Pull Request)別只停於上傳範例以及修改文件,以下我就使用我第一個貢獻的 LINE SDK(python) 帶大家了解開源專案會有的樣子。

Photo by tingyaoh
寫程式不管在哪個情境都會需要處理錯誤訊息,除了程式以外語言本身也需要這樣的功能,因此 python 就提供了一個標準函式庫 - atexit,這名字從 C 語言時態就存在的一個工具,主要是監聽程式當收到關閉請求時,可以先跳至已註冊的函式中先處理掉一些事情後再接著關閉服務,這個做法被稱為 Exit Handler,主要就是當程式收到關閉的訊號時最後會進入的函式們,一般較常見的會放上關閉資料庫連線、釋放記憶體、備份 cache 等等的功能,如此一來就能降低問題的產生,讓服務能夠可平滑的關閉、重啟。
不管使用 python 寫 腳本、後端、爬蟲 都可以使用 atexit 來處理例外錯誤,而我在 flask 中使用時與官方記載的註解有點不同,本篇就介紹一下 atexit 以及使用中不同的部分吧!
常常在寫程式時沒搞清楚作用域時很容易留坑或邏輯混亂,因此花時間理解每個語言的作用域也很重要。
之前因為專案有踩過一些坑,而大家在看套件時很能很常看到類似以下的 code:

約在一年前,當時公司的同事帶我玩過 Swagger,只是當時的我寫得不太習慣,直到寫這篇之前都還只是寫 markdown 來做 API 文件 😓,不過寫 API 到最後總是要開放給其他人需要測試以及用文件來描述功能,但只是寫 Readme.md 給其他人看感覺有點不負責任 (?),所以這個時候 Swagger 就發揮它的作用啦~
但是看著下面這張圖…
左邊那是怎樣!怎麼設定了一大堆右邊才出現一點點東西,估且不論還要設定 Model、Payload 什麼的,我想光寫左邊這份文件加測試的時間我都可以寫一份 Markdown 給其他同仁用 Postman 測試了…🤣🤣🤣
這時候套件就出動啦,這次要介紹的是 flask-restful-swagger-2.0,他是延續原版套件並加以更新的,不得不說再搭更新完後整體寫起來人性多了,像是文先寫法用 decorator 的寫法在路由之前、宣告 Schema 時只要用類別包起來之類的,讓寫文件可以用 python 的寫法去寫,除了特殊的設定需要比對 Swagger 官方文件外,在建一個 Swagger 格式的文件這樣已經超快了 💪
既然如此那就介紹一下他的相關設定並把它記錄下來吧!

對於 LINE Notify 其實早早就一直有在使用,不管是 server 健康檢查、空汙通知、直播主開台通知...等,都一直使用著,也是因為一直在用時有個困擾,就是每次都要找 API 文件並且看對應的參數...這在開發流程上浪費了不少時間,也因為到 Pypi 套件庫裡找 LINE Notify 相關套件時大多都只有推送功能,幾乎沒有對於認證流程的操作的 function ,文件也不齊全,在此同時也看到卡米哥的 GitHub 在此時上傳了一個 Notify SDK,畢竟我對 Ruby 還不算陌生,就直接參考卡米哥的架構並加上單元測試來完成本篇要介紹的 - Lotify

在 AWS 的 API gateway 上就有一個項目(Authorizer)是支援 Cognito 以及客制認證的方法,好處就是在進我們寫的 API 之前有個可以做身份確認的 Lambda Function 幫我們好前處理,而本篇則會介紹如何在 Serverless framework 上設定 API gateway 的 Authorizer 以及成功讓 API 回應,串接 Cognito 做使用部分就留給之後吧!
最近 LINE 釋出了一個我很喜歡的功能 - Icon Switch🎚,從字面上的意思就是可以切換機器人的 Icon (名稱以及大頭貼),過往開發者在使用時都只能較死板的使用同一個頭像或暱稱來回應使用者,而現在只要使用這個這個功能就可以很輕易地切換你想要的大頭貼以及暱稱了 😇
據可靠消息指出這以前是需要收費的功能,而現在則是免費推出來給大家使用,讓大家可以有更多的彈性去開發新功能~

最近在實作需要處理字串並且把它改成真實的判斷式,找 google 的過程中找到了一篇解答,當中剛好看到 Operator 這個套件,以下就用一些範例來示範~
處理的範例: 1=1AND2<5OR8>9
可以使用re把字串處理成陣列
re.split("AND|OR", payload)
# -> [1=1, 2<5, 8>9]
接著再使用
迴圈+同樣的方法去處理掉=、>、<
在處理完之後就可以找到所有的 key 以及 value 的陣列
接著就是要把 AND、OR 找出來存下來,這邊我會會使用split('AND', 1)只切第一個找到的AND字串,因為像是 SQL Where 子句的功能要一個一個處理才行,否則最後判斷會出問題。
while re.search('(AND)|(OR)', payload) is not None:
result = payload.split('AND', 1)
if result[0] and ... # 判斷 AND 字串
result = payload.split('OR', 1)
if result[0] and ... # 判斷 OR 字串
接著再切完之後會拿到operator_list知道裡面有哪些運算子的陣列,以原本的資料來說可以用迴圈去處理,這邊我就用簡單的範例來表示operator可以處理的東西。
像是operator除了可以判斷純數字以及純字串以外,也可判斷英文+數字的組合。
operator_list = ['>', '<', '=']
if operator_list[0] == '>':
result = operator.gt(2, 0)
if operator_list[1] == '<':
result = operator.lt('z01', 'z99')
if operator_list[2] == '=':
result = operator.eq('A', 'A')
使用以上方法可以比對運算子,而使用以下方法則可以比對AND以及OR
ex = ['AND', 'OR']
if ex[0] == 'AND':
result = operator.and_(1, 1)
if ex[1] == 'OR':
result = operator.or_(1, 0)
透過 operator 這個函示庫讓我處理運算子們可以少很多事,藉此紀錄一下工作上使遇到的一些方法以及問題,最驚訝的就是可以比對英文+中文的判斷式,以一些韌體的本版資訊來說他們會擁有這樣的組合,在上傳之後要比對時用這樣的方法就很方便,若有什麼問題歡迎留言給我 😊