声明性HTTP服务入口点。
py2http的Python项目详细描述
the_one_where_we_have_python_objects_and_we_want_to_make_an_http_service_out_of_them
py2http
将Python函数作为http服务进行调度。在
使用
从函数列表运行基本的HTTP服务。在
from py2http.service import run_http_service
# Define or import functions
def add(a, b):
return a + b
def multiply(a, b):
return a * b
class Divider:
def __init__(self, dividend):
self.dividend = divident
def divide(self, divisor):
return self.dividend / divisor
divider_from_ten = Divider(10)
# Make a list of functions or instance methods
func_list = [add, multiply, divider_from_ten.divide]
# Create an HTTP server
run_http_service(func_list)
默认情况下，服务器3030上的端口将侦听。在
^{pr2}$配置
run_http_service和其他入口点接受许多配置值来定制HTTP请求和响应的处理。配置文档列在文件config.yaml中。在
方法转换
通过展平init->；方法调用过程来公开类方法。在
from py2http import run_http_service
from py2http.decorators import mk_flat
class Adder:
def __init__(self, a):
self.a = a
def add(self, b):
return self.a + b
add = mk_flat(Adder, Adder.add, func_name='add')
func_list = [add]
run_http_service(func_list)
输入映射
默认情况下，服务器只接受JSON请求，并将请求体中的值作为关键字参数进行解析。您可以定义自定义输入映射器来对JSON主体或直接对HTTP库请求对象执行额外处理，例如默认注入或类型映射。在
from numpy import array
import soundfile
from py2http.decorators import handle_json_req
from my_service.data import fetch_user_data
@handle_json_req # extracts the JSON body and passes it to the input mapper as a dict
def array_input_mapper(input_kwargs):
return {'arr': array(input_kwargs.get('arr', [])),
'scalar': input_kwargs.get('scalar', 1)}
@handle_multipart_req # extracts a multipart body and passes it as a dict
def sound_file_input_mapper(input_kwargs):
file_input = input_kwargs['file_upload']
filename = file_input.filename
file_bytes = BytesIO(file_input.file)
wf, sr = soundfile.read(file_bytes)
return dict(input_kwargs, wf=wf, sr=sr, filename=filename)
def custom_header_input_mapper(req): # takes a raw aiohttp request
user = req.headers.get('User', '')
return {'user': user}
def array_multiplier(arr, scalar):
return arr * scalar
def save_audio(wf, sr, filename, storage_path='/audio'):
target = os.path.join(storage_path, filename)
soundfile.write(target, wf, sr)
def get_user_data(user):
return fetch_user_data(user)
array_multiplier.input_mapper = array_input_mapper
save_audio.input_mapper = sound_file_input_mapper
get_user_data.input_mapper = custom_header_input_mapper
输出映射
默认情况下，服务器将以JSON格式的HTTP响应发送handler函数的返回值。可以定义自定义输出映射器以执行额外的处理或类型转换。在
from io import BytesIO
import soundfile
from py2http.decorators import send_json_resp, send_html_resp
from my_service.data import fetch_blog
@send_json_resp # sends a JSON response
def array_output_mapper(output, input_kwargs):
return {'array_value': output['array_value'].tolist()}
@send_html_resp # sends an HTML response
def mk_html_list(output, input_kwargs):
tag = input_kwargs.get('tag', 'p')
item_list = output.split('\n')
result = [f'<{tag}>{item}</{tag}>' for item in item_list]
return ''.join(result)
def return_wav(output, input_kwargs):
wf, sr = output
wf_bytes = BytesIO()
soundfile.write(wf_bytes, wf, sr)
binary_result = wf_bytes.read()
return web.Response(body=binary_result, content_type='application/octet-stream')
def array_multiplier(arr, scalar):
return arr * scalar
def get_blog_posts(page):
return fetch_blog(page)
def download_audio(filename):
with open(filename) as fp:
wf, sr = soundfile.read(fp)
return wf, sr
array_multipler.output_mapper = array_output_mapper
get_blog_posts.output_mapper = mk_html_list
download_audio.output_mapper = return_wav
错误处理
托多
客户机生成
Py2http在运行之前为服务器生成一个OpenAPI规范。您可以将此文档与任何OpenAPI兼容的客户端工具一起使用。要提取规范，可以在运行之前生成一个服务器应用程序对象。在
from aiohttp import web
import json
from py2http import mk_http_service, run_aiohttp_service
def add(a, b):
return a + b
func_list = [add]
app = mk_http_service(func_list)
openapi_spec = app.openapi_spec
with open('~/openapi.json', 'w') as fp:
json.dump(openapi_spec, fp)
web.run_app(app, port=3030)
- 项目
标签：
欢迎加入QQ群-->： 979659372
