linker-atom 0.0.15

linker fastapi atom

项目开发文档

项目目录结构

├── api
│   ├── app.py                    FastApi实例(get_app)
│   ├── base.py                   封装FastApi类(UdfAPIRoute)
│   ├── dependencies.py           接口依赖注入
│   ├── route.py                  路由汇总
│   ├── interface                 接口路由目录
│   │   ├── healthcheck.py
│   ├── middleware                服务启动中间件
│   │   ├── event.py              事件中间件
│   │   ├── http.py               接口中间件
│   └── schema                    接口请求及返回数据模型
│       ├── payload.py
│       └── response.py
├── config                        
│   ├── __init__.py               配置文件
│   ├── logger.py                 日志配置
│   ├── serving.py                serving所需信息配置
│   └── skywalking.py             skywalking配置
├── dependency                    加密方法
│   └── model_protector.so        模型解密
├── lib                           公共方法封装目录
│   ├── common.py                 常用方法
│   ├── exception.py              异常类型
│   ├── load_image.py             加载图片
│   ├── log.py                    日志封装
│   ├── requests.py               requests请求参数/响应日志打印
│   └── share_memory.py           共享内存
├── README.md

项目介绍

• 项目基于fastapi进行二次封装, 加入特定的事件、路由封装、参数打印、接口耗时统计、异常捕获等
  • linker_atom.api.interface.healthcheck: 内置健康检查接口, 路由:$ATOM_API_PREFIX/v1/health/ping
  • linker_atom.api.middleware.event: 注册事件类型, 传递内部配置、按需启动skywalking、按需启动后台线程apscheduler、进行向Serving服务注册及定时发送心跳
  • linker_atom.api.middleware.http: 注册接口中间件, 进行全局异常捕获,返回标准结构数据
  • linker_atom.api.app: 获取fastapi app实例, get_app
  • linker_atom.api.base: 获取fastapi APIRoute实例, UdfAPIRoute
• 通用方法封装
  • linker_atom.lib.common: 异常方法捕获装饰器、长字符截断、限定内存类
  • linker_atom.lib.exception: 通用异常返回
  • linker_atom.lib.load_image: 根据src_type类型加载数据成numpy格式
  • linker_atom.lib.log: logging持久化封装、支持skywalking、日志格式化
  • linker_atom.lib.requests: requests封装、参数、返回数据、耗时日志打印
  • linker_atom.lib.share_memory: 共享内存类封装
• 加密方法目录
  • from linker_atom.dependency.model_protector import ModelProtector: 模型解密类

---

使用介绍

环境安装

pypi安装

pip install --no-cache-dir -U linker_atom -i https://pypi.org/simple 

wheel安装

pip install linker_atom-0.1.0-py3-none-any.whl --force-reinstall

使用示例

创建server.py

import os

import uvicorn
from uvicorn.loops.auto import auto_loop_setup

# 本地调试, 需在引用包之前, 设置环境变量DEBUG=true, 否则会一直向serving发送服务注册信息
os.environ.setdefault('DEBUG', 'true')
# Fastapi 实例
from linker_atom.api.app import get_app
# APIRoute 实例
from linker_atom.api.base import UdfAPIRoute
# 内置配置类实例
from linker_atom.config import settings
# 异常类
from linker_atom.lib.exception import VqlError
# 日志打印模块实例
from linker_atom.lib.log import logger

app = get_app()


@app.on_event("startup")
async def startup_event():
    # 算法实现,在服务启动前加载解密模型
    for model_id in settings.model_id_list:
        pass


# 实例化路由
route = UdfAPIRoute()


# 编写接口
@route.get('/error', name='error')
def error():
    raise VqlError(500)


@route.get('/normal', name='normal')
def normal():
    logger.info('ok!')
    return {'code': 0}


# 添加路由
app.include_router(
    router=route,
)


# 运行服务
def run():
    auto_loop_setup(True)
    uvicorn.run(
        app='server:app',
        host='0.0.0.0',
        port=settings.atom_port,
        workers=settings.atom_workers,
        access_log=False
    )


if __name__ == '__main__':
    run()

项目内置环境变量

变量名称	描述	默认值
ATOM_PORT	端口	8000
ATOM_WORKERS	worker数量	1
ATOM_PROCESS_URL	注册到Serving的处理接口	/atom/process
ATOM_API_PREFIX	API前缀	'/yolo'
ATOM_TITLE	标题	''
ATOM_DESC	描述	''
DEBUG	本地调试	False
LOG_BACKUP_COUNT	日志保留数量	30
LOG_DIR	日志子目录	atom
SW_SWITCH	skywalking是否开启	False
SW_AGENT_COLLECTOR_BACKEND_SERVICES	skywalking地址	''
SW_AGENT_NAME	skywalking服务名	''
SW_AGENT_INSTANCE_NAME	skywalking实例名称	uuid
SW_AGENT_LOG_REPORTER_ACTIVE	skywalking是否记录日志	True
SW_AGENT_LOG_REPORTER_LEVEL	skywalking日志等级	DEBUG
META	Serving传递信息json	''