Categories
程式開發

雲函數中使用Python-ORM: Peewee


前言

ORM(Object Ralational Mapping,對象關係映射)用來把對像模型表示的對象映射到基於SQL的關係模型數據庫結構中去。這樣,我們在具體的操作實體對象的時候,就不需要再去和復雜的SQL語句打交道,只需簡單的操作實體對象的屬性和方法。 ORM技術是在對象和關係之間提供了一條橋樑,前台的對象型數據和數據庫中的關係型的數據通過這個橋樑來相互轉化。

在傳統的開發過程中,無論是Java還是Python抑或Go,Nodejs等,都會有自己對應的一些工具,讓我們可以快速、簡單的使用數據庫。以Python為例,Django是常見的Python Web框架,其提供的ORM組件就是非常方便的。除此之外,Python語言還有很多常見的ORM工具,例如SQLObject、Storm、SQLAlchemy以及peewee。

本文將會通過在雲函數中使用peewee,來進行簡單的操作,將POST請求中的某些參數存儲到數據庫中,再讀取出來。

入門peewee

Peewee是一個簡單小巧的Python ORM,它非常容易學習,並且使用起來很直觀。在其官方文檔中,我們可以看到這樣一個Demo:

from peewee import *

db = SqliteDatabase('people.db')

class Person(Model):
name = CharField()
birthday = DateField()

class Meta:
database = db # This model uses the "people.db" database.

通過這樣一段代碼,我們可以基本明確幾件事:

使用的時候可以直接導入依賴:from peewee import *,需要先初始化數據庫鏈接:SqliteDatabase(‘people.db’),需要建立一個class用來描述結構,並且要繼承Model。

通過對文檔閱讀,我們可以看到peewee的增刪改查和Django所提供的ORM增刪改查很是相似。此處不進行更多說明。

本地調試

按照騰訊雲的Serverless分類下的雲函數格式要求,我們先完成一個基礎框架:

import json
from peewee import *

# 连接数据库

# 定义ServerlessFramework

def main_handler(event, context):
# 相关操作
return None

這裡需要注意,由於函數是無狀態的,且存在容器的複用,所以這裡鏈接數據庫的操作,要在入口方法外進行,這樣在一定程度下,可以提升性能和維持穩定。

方法內連接數據庫:函數每次被觸發(方法每次被調用)都會進行數據庫連接操作方法外連接數據庫:只會在容器啟動時進行初始化/數據庫連接,之後函數的每次觸發,都會復用當前的數據庫連接

為了在本地調試的更加愉快,此處增加一個新的test()方法:

def test():
event = {'body': json.dumps({"name": "hello world"}), 'headerParameters': {}, 'headers': {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'accept-encoding': 'gzip, deflate', 'accept-language': 'zh-CN,zh;q=0.9', 'cache-control': 'no-cache',
'connection': 'keep-alive', 'content-length': '27', 'content-type': 'application/x-www-form-urlencoded',
'cookie': 'Hm_lvt_a0c900918361b31d762d9cf4dc81ee5b=1574491278,1575257377', 'endpoint-timeout': '15',
'host': 'blog.0duzhan.com', 'origin': 'http://blog.0duzhan.com', 'pragma': 'no-cache',
'proxy-connection': 'keep-alive', 'referer': 'http://blog.0duzhan.com/admin/tag/new/?url=%2Fadmin%2Ftag%2F',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36',
'x-anonymous-consumer': 'true', 'x-api-requestid': '656622f3b008a0d406a376809b03b52c',
'x-b3-traceid': '656622f3b008a0d406a376809b03b52c', 'x-qualifier': '$LATEST'}, 'httpMethod': 'POST',
'path': '/admin/tag/new/', 'pathParameters': {}, 'queryString': {'url': '/admin/tag/'},
'queryStringParameters': {},
'requestContext': {'httpMethod': 'ANY', 'identity': {}, 'path': '/admin', 'serviceId': 'service-23ybmuq7',
'sourceIp': '119.123.224.87', 'stage': 'release'}}
print(main_handler(event, None))

if __name__ == "__main__":
test()

這樣做的目的是,可以直接通過執行該文件,模擬線上的API網關觸發器,例如當我的main_handler方法為以下代碼時:

def main_handler(event, context):
print("Event: ", event)

執行當前文件會輸出:

Event: {'body': '{"name": "hello world"}', 'headerParameters': {}, 'headers': {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9', 'accept-encoding': 'gzip, deflate', 'accept-language': 'zh-CN,zh;q=0.9', 'cache-control': 'no-cache', 'connection': 'keep-alive', 'content-length': '27', 'content-type': 'application/x-www-form-urlencoded', 'cookie': 'Hm_lvt_a0c900918361b31d762d9cf4dc81ee5b=1574491278,1575257377', 'endpoint-timeout': '15', 'host': 'blog.0duzhan.com', 'origin': 'http://blog.0duzhan.com', 'pragma': 'no-cache', 'proxy-connection': 'keep-alive', 'referer': 'http://blog.0duzhan.com/admin/tag/new/?url=%2Fadmin%2Ftag%2F', 'upgrade-insecure-requests': '1', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36', 'x-anonymous-consumer': 'true', 'x-api-requestid': '656622f3b008a0d406a376809b03b52c', 'x-b3-traceid': '656622f3b008a0d406a376809b03b52c', 'x-qualifier': '$LATEST'}, 'httpMethod': 'POST', 'path': '/admin/tag/new/', 'pathParameters': {}, 'queryString': {'url': '/admin/tag/'}, 'queryStringParameters': {}, 'requestContext': {'httpMethod': 'ANY', 'identity': {}, 'path': '/admin', 'serviceId': 'service-23ybmuq7', 'sourceIp': '119.123.224.87', 'stage': 'release'}}
None

此時我們建立一個數據庫:

雲函數中使用Python-ORM: Peewee 1

並且完成代碼:

import json
from peewee import *

# 连接数据库
database = MySQLDatabase('数据库名', user='数据库用户', host='数据库Host', port=数据库端口,
password="数据库密码")

# 定义ServerlessFramework
class ServerlessFramework(Model):
componentName = CharField()

class Meta:
database = database

# 建立表(此处推荐不在函数中建立,最好初始化之后,屏蔽掉该语句)
# ServerlessFramework.create_table()

def main_handler(event, context):
print("Event: ", event)

body = json.loads(event["body"])
print("Body: ", body)

try:
ServerlessFramework(componentName=body["name"]).save()
return [eve.componentName for eve in ServerlessFramework.select()]
except Exception as e:
return str(e)

def test():
event = {'body': json.dumps({"name": "hello world"}), 'headerParameters': {}, 'headers': {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'accept-encoding': 'gzip, deflate', 'accept-language': 'zh-CN,zh;q=0.9', 'cache-control': 'no-cache',
'connection': 'keep-alive', 'content-length': '27', 'content-type': 'application/x-www-form-urlencoded',
'cookie': 'Hm_lvt_a0c900918361b31d762d9cf4dc81ee5b=1574491278,1575257377', 'endpoint-timeout': '15',
'host': 'blog.0duzhan.com', 'origin': 'http://blog.0duzhan.com', 'pragma': 'no-cache',
'proxy-connection': 'keep-alive', 'referer': 'http://blog.0duzhan.com/admin/tag/new/?url=%2Fadmin%2Ftag%2F',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36',
'x-anonymous-consumer': 'true', 'x-api-requestid': '656622f3b008a0d406a376809b03b52c',
'x-b3-traceid': '656622f3b008a0d406a376809b03b52c', 'x-qualifier': '$LATEST'}, 'httpMethod': 'POST',
'path': '/admin/tag/new/', 'pathParameters': {}, 'queryString': {'url': '/admin/tag/'},
'queryStringParameters': {},
'requestContext': {'httpMethod': 'ANY', 'identity': {}, 'path': '/admin', 'serviceId': 'service-23ybmuq7',
'sourceIp': '119.123.224.87', 'stage': 'release'}}
print(main_handler(event, None))

if __name__ == "__main__":
test()

此時我們本地執行可以得到結果:

Event: {'body': '{"name": "hello world"}', 'headerParameters': {}, 'headers': {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9', 'accept-encoding': 'gzip, deflate', 'accept-language': 'zh-CN,zh;q=0.9', 'cache-control': 'no-cache', 'connection': 'keep-alive', 'content-length': '27', 'content-type': 'application/x-www-form-urlencoded', 'cookie': 'Hm_lvt_a0c900918361b31d762d9cf4dc81ee5b=1574491278,1575257377', 'endpoint-timeout': '15', 'host': 'blog.0duzhan.com', 'origin': 'http://blog.0duzhan.com', 'pragma': 'no-cache', 'proxy-connection': 'keep-alive', 'referer': 'http://blog.0duzhan.com/admin/tag/new/?url=%2Fadmin%2Ftag%2F', 'upgrade-insecure-requests': '1', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36', 'x-anonymous-consumer': 'true', 'x-api-requestid': '656622f3b008a0d406a376809b03b52c', 'x-b3-traceid': '656622f3b008a0d406a376809b03b52c', 'x-qualifier': '$LATEST'}, 'httpMethod': 'POST', 'path': '/admin/tag/new/', 'pathParameters': {}, 'queryString': {'url': '/admin/tag/'}, 'queryStringParameters': {}, 'requestContext': {'httpMethod': 'ANY', 'identity': {}, 'path': '/admin', 'serviceId': 'service-23ybmuq7', 'sourceIp': '119.123.224.87', 'stage': 'release'}}
Body: {'name': 'hello world'}
['test', 'test2', 'test2', 'hello world']

可以看到,我們的程序是沒問題,數據也可以成功被寫入到數據庫:

雲函數中使用Python-ORM: Peewee 2

部署到線上

部署到線上,我們可以將數據庫所在子網和雲函數所在子網設置成一致,通過內網直接使用:

數據庫子網:

雲函數中使用Python-ORM: Peewee 3

雲函數VPC:

雲函數中使用Python-ORM: Peewee 4

完成之後,我們需要在項目當前目錄下安裝相關依賴:pip3 install peewee -t ./

同時我們需要將數據的鏈接地址,修改成內網地址,最後上傳代碼包函數:

雲函數中使用Python-ORM: Peewee 5

通過PostMan進行測試:

雲函數中使用Python-ORM: Peewee 6

當然,如果想通過Serverless命令行工具部署也是比較容易的:

MyPeeweeTest:
component: "@serverless/tencent-scf"
inputs:
name: mypeewee_test
codeUri: ./src
handler: index.main_handler
runtime: Python3.6
region: ap-guangzhou
vpcConfig:
subnetId: 'subnet-j2ga1kew'
vpcId: 'vpc-h3933w5f'
events:
- apigw:
name: serverless_test
parameters:
serviceId: service-cyjmc4eg
protocols:
- http
description: the serverless service
environment: release
endpoints:
- path: /users
method: POST

總結

至此,我們完成了在雲函數中使用peewee的Demo,這個Demo還有很多可以優化的地方,例如將數據庫等信息寫入到環境變量中,這樣便於我們更新維護等。

雲函數其實是可以使用絕大部分框架和依賴的,就看我們如何去使用,當然這個組件是比較簡單的,不需要進行改造,有一些組件就算不能用,也是可以根據一些改造,將其跑在雲函數中。