在Python中使用API创建Google Cloud Function [英] Create Google Cloud Function using API in Python
问题描述
我正在使用Python（3.6）& Django（1.10），其中我需要使用API请求在Google云中创建一个函数。
如何在创建zip格式的同时上传代码这个函数？ 这是我尝试过的： From views.py ： 更新后的代码如下：
def post（self，request，* args，** kwargs）：
if request .method =='POST'：
post_data = request.POST.copy（）
post_data.update（{'user'：request.user.pk}）
form = forms.SlsForm （post_data，request.FILES）
print（'get post request'）
if form.is_valid（）：
func_obj = form
func_obj.user = request.user
func_obj.project = form.cleaned_data ['project']
func_obj.fname = form.cleaned_data ['fname']
func_obj.fmemory = form.cleaned_data ['fmemory']
func_obj.entryPoint = form.cleaned_data ['entry Point']
func_obj.sourceFile = form.cleaned_data ['sourceFile']
func_obj.sc_github = form.cleaned_data ['sc_github']
func_obj.sc_inline_index = form.cleaned_data ['sc_inline_index' ]
func_obj.sc_inline_package = form.cleaned_data ['sc_inline_package']
func_obj.bucket = form.cleaned_data ['bucket']
func_obj.save（）
service = discovery。 build（'cloudfunctions'，'v1'，http = views.getauth（），cache_discovery = False）
requ = service.projects（）。locations（）。functions（）。generateUploadUrl（parent ='projects /' + func_obj.project +'/ locations / us-central1'，body = {}）
resp = requ.execute（）
print（resp）
try：
auth = views.getauth（）
＃准备请求正文
req_body = {
CloudFunction：{
name：func_obj.fname，
entryPoint：func_obj.entryPoint，
timeout：'60s'，
availableMemoryMb：func_obj.fmemory，
sourceArchiveUrl ：func_obj.sc_github，
}，
sourceUploadUrl：func_obj.bucket，
}
service = discovery.build（'cloudfunctions'，'v1beta2'，http = auth ，cachce_dicovery = False）
func_req = service.projects（）。locations（）。functions（）。create（location ='projects /'+ func_obj.project
+'/ locations / - '，
body = req_body）
func_res = func_req.execute（）
print（func_res）
return HttpResponse（'Submitted'，）
除了：
return HttpResponse（status = 500）
return HttpResponse（'Sent！'）
if form.is_valid（）：
func_obj = form
func_obj.user = request.user
func_obj.project = form.cleaned_data ['project']
func_obj.fname = form.cleaned_data ['fname']
func_obj.fmemory = form.cleaned_data ['fmemory']
func_obj.entryPoint = form.cleaned_data ['entryPoint']
func_obj.sourceFile = form。 clean_data ['sourceFile']
func_obj.sc_github = form.cleaned_data ['sc_github']
func_obj.sc_inline_index = form.cleaned_data ['sc_inline_index']
func_obj.sc_inline_package = form.cleaned_data [ 'sc_inline_package']
func_obj.bucket = form.clean ed_data ['bucket']
func_obj.save（）
########################### ############################################
＃FIRST使用存储桶创建功能的方法
####################################### ################################
file_name = os.path.join（IGui。 （'cloudfunctions'，'v1'）
$ b func_api = service.projects（）。locations（）。functions（）
url_svc_req = func_api.generateUploadUrl（parent ='projects /'
+ func_obj.project
+'/ locations / us -central1'，
body = {}）
url_svc_res = url_svc_req.execu te（）
print（url_svc_res）
$ b upload_url = url_svc_res ['uploadUrl']
print（upload_url）
headers = {
'content-type '：'application / zip'，
'x-goog-content-length-range'：'0,104857600'
}
print（requests.put（upload_url，headers = headers， data = func_obj.sourceFile.name））
auth = views.getauth（）
＃准备请求正文
name =projects / {} / locations / us-central1 / functions / {} .format（func_obj.project，func_obj.fname）
print（name）
req_body = {
name：name，
entryPoint：func_obj.entryPoint，
timeout：3.5s，
availableMemoryMb：func_obj.fmemory，
sourceUploadUrl：upload_url，
httpsTrigger：{}，
}
服务=光盘overy.build（'cloudfunctions'，'v1'）
func_api = service.projects（）。locations（）。functions（）
response = func_api.create（location ='projects / '+ func_obj.project +'/ locations / us-central1'，
body = req_body）.execute（）
pprint.pprint（response）
$ b> / p>
upload_url = url_svc_res ['uploadUrl']
print（upload_url）
headers = {
'content-type'：'application / zip'，
'x-goog-content-length-range'：'0,104857600'
}
print（requests.put（upload_url， headers = header，data = func_obj.sourceFile.name））
在你有请求内的字典CloudFunction请求体。
request_body = {
name：parent +' / functions /'+ name，
entryPoint：entry_point，
sourceUploadUrl：upload_url，
httpsTrigger：{}
}
我推荐使用试试这个API来发现projects.locations.functions.create的结构。
sourceArchiveUrl和sourceUploadUrl不能同时出现。这在 Resorce Cloud Function
// Union字段source_code只能为以下值之一：
sourceArchiveUrl：string ，
sourceRepository：{object（SourceRepository）}，
sourceUploadUrl：string，
//联合字段source_code的可能类型列表的结束。
在其余的答案中，我假设您要使用sourceUploadUrl 。它要求您通过 .generateUploadUrl（...）。execute（）向您返回一个URL。请参阅文档 ：
sourceUploadUrl - > 字符串
由[google.cloud.functions.v1。 GenerateUploadUrl ] []
生成的用于源上传的Google云端存储签名网址
但是在传递它之前，您需要上传一个zip文件到这个URL： code> curl -X PUT$ {URL}-H'content-type：application / zip'-H'x-goog-content-length-range：0,104857600'-T test.zip
或在python中：
< code'headers = {
'content-type'：'application / zip'，
'x-goog-content-length-range'：'0,104857600'
}
print（requests.put（upload_url，headers = headers，data = data））
这是最棘手的部分： 案件很重要，它应该是小写字母。由于签名是通过散列计算的（这里） 你需要'content-type'：'application / zip'。我逻辑推导出这个，因为文档没有提到它。 （此处） 104857600，以前的最大值是一个神奇的数字，我没有在任何地方发现过。
x-goog-content-length-range：min，max 对所有 PUT 云存储请求，并假定在这种情况下是隐含的。更多内容此处
其中 data 是一个FileLikeObject。
我还假设您要使用 httpsTrigger 。对于云端功能，您只能选择一个触发字段。 Here 据说触发器是联盟领域。但对于httpsTrigger，您可以将其保留为空字典，因为其内容不会影响结果。
request_body = {
name：parent +'/ functions /'+ name，
entryPoint：entry_point，
sourceUploadUrl：upload_url，
httpsTrigger：{}
}
对于 .create（），您可以安全地使用'v1'而不是'v1beta2'。
以下是一个完整的工作示例。如果我把它作为你的代码的一部分呈现给你，这将会变得复杂，但是你可以很容易地将它集成到其中。
import pprint
导入zipfile
导入请求
从tempfile导入TemporaryFile
from googleapiclient导入发现
project_id ='your_project_id'
region ='us- （project_id，region）
print（parent）
name ='ExampleFunctionFibonacci'
entry_point =中心1'
父='projects / {} / locations / {} fibonacci
service = discovery.build（'cloudfunctions'，'v1'）
CloudFunctionsAPI = service.projects（）。locations（）。functions（）$ b $ upload_url = CloudFunctionsAPI .generateUploadUrl（parent = parent，body = {}）。execute（）['uploadUrl']
print（upload_url）
payload =/ ** $
*
* @param {Object} req云功能请求上下文
* @param {b $ b *}响应任何可以在主体中提供消息字段的HTTP请求。对象} res Cloud函数响应上下文
* /
exports.fib = function+ entry_point +（req，res）{
if（req.body.message === undefined） {
//这是一个错误情况，因为需要消息
res.status（400）.send（'No message defined！'）;
} else {
//一切正常
console.log（req.body.message）;
res.status（200）.end（）;
}
};
与TemporaryFile（）作为数据：
with zipfile.ZipFile（data，'w'，zipfile。 ZIP_DEFLATED）作为压缩文件：
archive.writestr（'function.js'，payload）
data.seek（0）
headers = {
'content-type '：'application / zip'，
'x-goog-content-length-range'：'0,104857600'
}
print（requests.put（upload_url，headers = headers， data = data））
＃准备请求正文
＃https://cloud.google.com/functions/docs/reference/rest/v1/projects.locations.functions#resource- cloudfunction
$ b request_body = {
name：parent +'/ functions /'+ name，
entryPoint：entry_point，
sourceUploadUrl：upload_url，
httpsTrigger：{}
}
print（'https：// [YOUR_REGION] - [YOUR_PROJECT_ID] .cloudfunctions.net / [NAME_OF_THE_TRIGGER]'.format）
response = CloudFunctionsAPI.create（location = parent，body = request_body）.execute（）
pprint.pprint（响应）
打开并上传一个zip文件，如下所示： p>
file_name = os.path.join（IGui.settings.BASE_DIR，'media / archives /'，func_obj.sourceFile.name）
headers = {
'content-type'：'application / zip'，
'x-goog-content-length-range'：'0,104857600'
}
打开（file_name，'rb'）作为数据：
print（requests.put（upload_url，headers = headers，data = data））
I'm working on a project with Python(3.6) & Django(1.10) in which I need to create a function at Google cloud using API request.
How can upload code in the form of a zip archive while creating that function?
Here's what I have tried:
From views.py :
def post(self, request, *args, **kwargs):
if request.method == 'POST':
post_data = request.POST.copy()
post_data.update({'user': request.user.pk})
form = forms.SlsForm(post_data, request.FILES)
print('get post request')
if form.is_valid():
func_obj = form
func_obj.user = request.user
func_obj.project = form.cleaned_data['project']
func_obj.fname = form.cleaned_data['fname']
func_obj.fmemory = form.cleaned_data['fmemory']
func_obj.entryPoint = form.cleaned_data['entryPoint']
func_obj.sourceFile = form.cleaned_data['sourceFile']
func_obj.sc_github = form.cleaned_data['sc_github']
func_obj.sc_inline_index = form.cleaned_data['sc_inline_index']
func_obj.sc_inline_package = form.cleaned_data['sc_inline_package']
func_obj.bucket = form.cleaned_data['bucket']
func_obj.save()
service = discovery.build('cloudfunctions', 'v1', http=views.getauth(), cache_discovery=False)
requ = service.projects().locations().functions().generateUploadUrl(parent='projects/' + func_obj.project + '/locations/us-central1', body={})
resp = requ.execute()
print(resp)
try:
auth = views.getauth()
# Prepare Request Body
req_body = {
"CloudFunction": {
"name": func_obj.fname,
"entryPoint": func_obj.entryPoint,
"timeout": '60s',
"availableMemoryMb": func_obj.fmemory,
"sourceArchiveUrl": func_obj.sc_github,
},
"sourceUploadUrl": func_obj.bucket,
}
service = discovery.build('cloudfunctions', 'v1beta2', http=auth, cachce_dicovery=False)
func_req = service.projects().locations().functions().create(location='projects/' + func_obj.project
+ '/locations/-',
body=req_body)
func_res = func_req.execute()
print(func_res)
return HttpResponse('Submitted',)
except:
return HttpResponse(status=500)
return HttpResponse('Sent!')
Updated Code below:
if form.is_valid():
func_obj = form
func_obj.user = request.user
func_obj.project = form.cleaned_data['project']
func_obj.fname = form.cleaned_data['fname']
func_obj.fmemory = form.cleaned_data['fmemory']
func_obj.entryPoint = form.cleaned_data['entryPoint']
func_obj.sourceFile = form.cleaned_data['sourceFile']
func_obj.sc_github = form.cleaned_data['sc_github']
func_obj.sc_inline_index = form.cleaned_data['sc_inline_index']
func_obj.sc_inline_package = form.cleaned_data['sc_inline_package']
func_obj.bucket = form.cleaned_data['bucket']
func_obj.save()
#######################################################################
# FIRST APPROACH FOR FUNCTION CREATION USING STORAGE BUCKET
#######################################################################
file_name = os.path.join(IGui.settings.BASE_DIR, 'media/archives/', func_obj.sourceFile.name)
print(file_name)
service = discovery.build('cloudfunctions', 'v1')
func_api = service.projects().locations().functions()
url_svc_req = func_api.generateUploadUrl(parent='projects/'
+ func_obj.project
+ '/locations/us-central1',
body={})
url_svc_res = url_svc_req.execute()
print(url_svc_res)
upload_url = url_svc_res['uploadUrl']
print(upload_url)
headers = {
'content-type': 'application/zip',
'x-goog-content-length-range': '0,104857600'
}
print(requests.put(upload_url, headers=headers, data=func_obj.sourceFile.name))
auth = views.getauth()
# Prepare Request Body
name = "projects/{}/locations/us-central1/functions/{}".format(func_obj.project, func_obj.fname,)
print(name)
req_body = {
"name": name,
"entryPoint": func_obj.entryPoint,
"timeout": "3.5s",
"availableMemoryMb": func_obj.fmemory,
"sourceUploadUrl": upload_url,
"httpsTrigger": {},
}
service = discovery.build('cloudfunctions', 'v1')
func_api = service.projects().locations().functions()
response = func_api.create(location='projects/' + func_obj.project + '/locations/us-central1',
body=req_body).execute()
pprint.pprint(response)
Now the function has been created successfully, but it fails because the source code doesn't upload to storage bucket, that's maybe something wrong at:
upload_url = url_svc_res['uploadUrl']
print(upload_url)
headers = {
'content-type': 'application/zip',
'x-goog-content-length-range': '0,104857600'
}
print(requests.put(upload_url, headers=headers, data=func_obj.sourceFile.name))
In the request body you have a dictionary "CloudFunction" inside the request. The content of "CloudFunction" should be directly in request.
request_body = {
"name": parent + '/functions/' + name,
"entryPoint": entry_point,
"sourceUploadUrl": upload_url,
"httpsTrigger": {}
}
I recomend using "Try this API" to discover the structure of projects.locations.functions.create .
"sourceArchiveUrl" and "sourceUploadUrl" can't appear together. This is explained in Resorce Cloud Function:
// Union field source_code can be only one of the following:
"sourceArchiveUrl": string,
"sourceRepository": { object(SourceRepository) },
"sourceUploadUrl": string,
// End of list of possible types for union field source_code.
In the rest of the answer I assume that you want to use "sourceUploadUrl". It requires you to pass it a URL returned to you by .generateUploadUrl(...).execute(). See documentation:
sourceUploadUrl -> string
The Google Cloud Storage signed URL used for source uploading, generated by [google.cloud.functions.v1.GenerateUploadUrl][]
But before passing it you need to upload a zip file to this URL:
curl -X PUT "${URL}" -H 'content-type:application/zip' -H 'x-goog-content-length-range: 0,104857600' -T test.zip
or in python:
headers = {
'content-type':'application/zip',
'x-goog-content-length-range':'0,104857600'
}
print(requests.put(upload_url, headers=headers, data=data))
This is the trickiest part:
the case matters and it should be lowercase. Because the signature is calculated from a hash (here)
you need 'content-type':'application/zip'. I deduced this one logically, because documentation doesn't mention it. (here)
x-goog-content-length-range: min,maxis obligatory for allPUTrequests for cloud storage and is assumed implicitly in this case. More on it here104857600, the max in previous entry, is a magical number which I didn't found mentioned anywhere.
where data is a FileLikeObject.
I also assume that you want to use the httpsTrigger. For a cloud function you can only choose one trigger field. Here it's said that trigger is a Union field. For httpsTrigger however that you can just leave it to be an empty dictionary, as its content do not affect the outcome. As of now.
request_body = {
"name": parent + '/functions/' + name,
"entryPoint": entry_point,
"sourceUploadUrl": upload_url,
"httpsTrigger": {}
}
You can safely use 'v1' instead of 'v1beta2' for .create().
Here is a full working example. It would be to complicated if I presented it to you as part of your code, but you can easily integrate it.
import pprint
import zipfile
import requests
from tempfile import TemporaryFile
from googleapiclient import discovery
project_id = 'your_project_id'
region = 'us-central1'
parent = 'projects/{}/locations/{}'.format(project_id, region)
print(parent)
name = 'ExampleFunctionFibonacci'
entry_point = "fibonacci"
service = discovery.build('cloudfunctions', 'v1')
CloudFunctionsAPI = service.projects().locations().functions()
upload_url = CloudFunctionsAPI.generateUploadUrl(parent=parent, body={}).execute()['uploadUrl']
print(upload_url)
payload = """/**
* Responds to any HTTP request that can provide a "message" field in the body.
*
* @param {Object} req Cloud Function request context.
* @param {Object} res Cloud Function response context.
*/
exports.fib = function """ + entry_point + """ (req, res) {
if (req.body.message === undefined) {
// This is an error case, as "message" is required
res.status(400).send('No message defined!');
} else {
// Everything is ok
console.log(req.body.message);
res.status(200).end();
}
};"""
with TemporaryFile() as data:
with zipfile.ZipFile(data, 'w', zipfile.ZIP_DEFLATED) as archive:
archive.writestr('function.js', payload)
data.seek(0)
headers = {
'content-type':'application/zip',
'x-goog-content-length-range':'0,104857600'
}
print(requests.put(upload_url, headers=headers, data=data))
# Prepare Request Body
# https://cloud.google.com/functions/docs/reference/rest/v1/projects.locations.functions#resource-cloudfunction
request_body = {
"name": parent + '/functions/' + name,
"entryPoint": entry_point,
"sourceUploadUrl": upload_url,
"httpsTrigger": {}
}
print('https://[YOUR_REGION]-[YOUR_PROJECT_ID].cloudfunctions.net/[NAME_OF_THE_TRIGGER]'.format)
response = CloudFunctionsAPI.create(location=parent, body=request_body).execute()
pprint.pprint(response)
Open and upload a zip file like following:
file_name = os.path.join(IGui.settings.BASE_DIR, 'media/archives/', func_obj.sourceFile.name)
headers = {
'content-type': 'application/zip',
'x-goog-content-length-range': '0,104857600'
}
with open(file_name, 'rb') as data:
print(requests.put(upload_url, headers=headers, data=data))
这篇关于在Python中使用API创建Google Cloud Function的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
