This repository has been archived by the owner on Jan 3, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 44
/
runserver.py
76 lines (63 loc) · 3.2 KB
/
runserver.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# # /ai4e_api_tools has been added to the PYTHONPATH, so we can reference those libraries directly.
from time import sleep
import json
from flask import Flask, request, abort
from ai4e_app_insights_wrapper import AI4EAppInsights
from ai4e_service import APIService
print('Creating Application')
app = Flask(__name__)
# Use the AI4EAppInsights library to send log messages. NOT REQUIRED
log = AI4EAppInsights()
# Use the APIService to executes your functions within a logging trace, supports long-running/async functions,
# handles SIGTERM signals from AKS, etc., and handles concurrent requests.
with app.app_context():
ai4e_service = APIService(app, log)
# Define a function for processing request data, if applicable. This function loads data or files into
# a dictionary for access in your API function. We pass this function as a parameter to your API setup.
def process_request_data(request):
return_values = {'data': None}
try:
# Attempt to load the body
return_values['data'] = request.get_json()
except:
log.log_error('Unable to load the request data') # Log to Application Insights
return return_values
# Define a function that runs your model. This could be in a library.
def run_model(taskId, body):
# Update the task status, so the caller knows it has been accepted and is running.
ai4e_service.api_task_manager.UpdateTaskStatus(taskId, 'running model')
log.log_debug('Running model', taskId) # Log to Application Insights
#INSERT_YOUR_MODEL_CALL_HERE
sleep(10) # replace with real code
# POST, long-running/async API endpoint example
@ai4e_service.api_async_func(
api_path = '/example',
methods = ['POST'],
request_processing_function = process_request_data, # This is the data process function that you created above.
maximum_concurrent_requests = 3, # If the number of requests exceed this limit, a 503 is returned to the caller.
content_types = ['application/json'],
content_max_length = 1000, # In bytes
trace_name = 'post:my_long_running_funct')
def default_post(*args, **kwargs):
# Since this is an async function, we need to keep the task updated.
taskId = kwargs.get('taskId')
log.log_debug('Started task', taskId) # Log to Application Insights
# Get the data from the dictionary key that you assigned in your process_request_data function.
request_json = kwargs.get('data')
if not request_json:
ai4e_service.api_task_manager.FailTask(taskId, 'Task failed - Body was empty or could not be parsed.')
return -1
# Run your model function
run_model(taskId, request_json)
# Once complete, ensure the status is updated.
log.log_debug('Completed task', taskId) # Log to Application Insights
# Update the task with a completion event.
ai4e_service.api_task_manager.CompleteTask(taskId, 'completed')
# GET, sync API endpoint example
@ai4e_service.api_sync_func(api_path = '/echo/<string:text>', methods = ['GET'], maximum_concurrent_requests = 1000, trace_name = 'get:echo', kwargs = {'text'})
def echo(*args, **kwargs):
return 'Echo: ' + kwargs['text']
if __name__ == '__main__':
app.run()