在Python中使用API​​创建Google Cloud Function

本文介绍了在Python中使用API​​创建Google Cloud Function的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述 29岁程序员，3月因学历无情被辞！ 我正在使用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。请参阅文档 ： 但是在传递它之前，您需要上传一个zip文件到这个URL： code> curl -X PUT$ {URL}-H'content-type：application / zip'-H'x-goog-content-length-range：0,104857600'-T test.zip 或在python中： 'content-type'：'application / zip'，'x-goog-content-length-range'：'0,104857600'} print（requests.put（upload_url，headers = headers，data = data）） 这是最棘手的部分： 案件很重要，它应该是小写字母。由于签名是通过散列计算的（这里） 你需要'content-type'：'application / zip'。我逻辑推导出这个，因为文档没有提到它。 （此处） x-goog-content-length-range：min，max 对所有 PUT 云存储请求，并假定在这种情况下是隐含的。更多内容此处 104857600，以前的最大值是一个神奇的数字，我没有在任何地方发现过。 其中 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!') 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: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.zipor 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,max is obligatory for all PUT requests 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 pprintimport zipfileimport requestsfrom tempfile import TemporaryFilefrom googleapiclient import discoveryproject_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-cloudfunctionrequest_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的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！ 上岸，阿里云！