推理服务部署

通过 get_deployment_detail 可以快速返回一个deployment对象,而不必直接实例化 Deployment(...)

调用服务: 
        deploy = Deployment.get_service(200)
        print(deploy.call_service({}))
---
获取服务列表: 
        deploys = Deployment.list_deployments(3976)
        for deploy in deploys:
            detail = deploy.get_deployment_detail()
        print(deploys)
---
修复服务名称:
        deploy = Deployment.rename_deployment(deploy_id, "test")

Deployment (APIObject) dataclass

Deployment(id: int = 0, project_id: int = 0, user_id: int = 0, name: str = '', service_name: str = '', model_inst_id: int = 0, serverless_infer_id: int = 0, deploy_model: str = '', route_path: str = '', token: str = '', token_url: str = '', status: int = 0, infer_lock: int = 0, min_pod: int = 0, max_pod: int = 0, create_time: str = '', update_time: str = '', deploy_time: str = '', is_del: int = 0)

call_service(self, data=None)

调用服务

Parameters:
  • data (Any) – 服务调用入参

Returns:
  • Any – 响应body

Source code in deepwisdom/models/deployment.py
def call_service(self, data: Any = None) -> Any:
    """调用服务

    Args:
        data (Any): 服务调用入参

    Returns:
        Any: 响应body
    """
    if data is None:
        api_desc = self.get_service_api(self.id)
        data = api_desc.post_example

    header = {
        "Authorization": "Bearer  {}".format(self.token),
        "Content-Type": "application/json; charset=utf-8",
    }
    rsp = self._client.raw_request(
        "post", self.token_url, json=data, headers=header)
    return rsp

create(project=None, name='Untitled Service', gpu_mem=2, memory_limit=3, max_pod=1, min_pod=1) classmethod

创建服务

Parameters:
  • project (Project) – 项目对象

  • name (str) – 名称

  • gpu_mem (int) – gpu显存限制GB

  • memory_limit (int) – GB 内存限制

  • max_pod (int) – 最大pod数

  • min_pod (int) – 最小pod数

Source code in deepwisdom/models/deployment.py
@classmethod
def create(cls, project: Optional[Project] = None, name: str = "Untitled Service", gpu_mem: int = 2,
           memory_limit: int = 3, max_pod: int = 1, min_pod: int = 1) -> "Deployment":
    """
    创建服务
    Args:
        project (Project): 项目对象
        name (str): 名称
        gpu_mem (int): gpu显存限制GB
        memory_limit (int): GB 内存限制
        max_pod (int): 最大pod数
        min_pod (int):  最小pod数

    Returns:

    """
    req = CreateDeploymentRequest(project_id=project.project_id, model_inst_id=project.recommended_model().model_id,
                                  name=name, gpu_mem=gpu_mem, memory_limit=memory_limit, max_pod=max_pod,
                                  min_pod=min_pod)

    return cls.create_from_req(req)

create_from_req(options) classmethod

创建服务

Parameters:
  • options (CreateDeploymentRequest) – 服务创建请求参数

Returns:
  • [Deployment] – 返回具体的服务对象

Source code in deepwisdom/models/deployment.py
@classmethod
def create_from_req(cls, options: CreateDeploymentRequest) -> "Deployment":
    """创建服务

    Args:
        options (CreateDeploymentRequest): 服务创建请求参数

    Returns:
        [Deployment]: 返回具体的服务对象
    """
    data = {}
    data.update(options)
    rsp = cls._client._post(API_URL.DEPLOY_CREATE_SERVICE, data)
    if "data" in rsp and "ret" in rsp["data"] and len(rsp["data"]["ret"]) == 1:
        deployment = cls.from_server_data(rsp['data']["ret"][0])
        # 等待服务部署完成
        while True:
            status = deployment.get_service_status()
            if status == SVR_STATUS.FAIL:
                raise err.CreateDeploymentError
            if status == SVR_STATUS.RUNNING:
                return deployment
    else:
        raise err.CreateDeploymentError

delete_deployments(service_ids) classmethod

删除服务

Parameters:
  • service_ids (int64) – 服务id

Source code in deepwisdom/models/deployment.py
@classmethod
def delete_deployments(cls, service_ids: List[int]) -> bool:
    """删除服务

    Args:
        service_ids (int64): 服务id
    """
    data = {
        "svc_ids": service_ids
    }
    rsp = cls._client._delete(API_URL.DEPLOY_DELETE_DEPLOYMENT, data)
    if rsp['code'] == 200 and rsp['message'] == 'ok':
        return True
    return False

get_deployment_log(svc_id) classmethod

获取服务日志

Parameters:
  • svc_id (int) – 服务id

Returns:
  • str – 日志内容

Source code in deepwisdom/models/deployment.py
@classmethod
def get_deployment_log(cls, svc_id: int) -> str:
    """获取服务日志

    Args:
        svc_id (int): 服务id

    Returns:
        str: 日志内容
    """
    data = {
        "service_id": svc_id
    }
    rsp = cls._client._get(API_URL.DEPLOY_GET_DEPLOYMENT_LOG, data)
    if "data" in rsp:
        return rsp['data']
    return None

get_service(service_id, **kwargs) classmethod

获取服务详情

Parameters:
  • service_id (int) – 服务id

Returns:
  • Deployment – 服务详情

Source code in deepwisdom/models/deployment.py
@classmethod
def get_service(cls, service_id, **kwargs) -> "Deployment":
    """获取服务详情

    Args:
        service_id (int): 服务id
    Returns:
        Deployment: 服务详情
    """
    data = {
        "service_id": service_id
    }
    rsp = cls._server_data(API_URL.DEPLOY_GET_SERVICE_DETAIL, data)
    return cls.from_server_data(rsp)

get_service_api(svc_id) classmethod

获取服务api

Parameters:
  • svc_id (int) – 服务id

Returns:
  • ServiceApiInfo – api详情

Source code in deepwisdom/models/deployment.py
@classmethod
def get_service_api(cls, svc_id: int) -> ServiceApiInfo:
    """获取服务api

    Args:
        svc_id (int): 服务id

    Returns:
        ServiceApiInfo: api详情
    """
    data = {
        "svc_id": svc_id
    }
    rsp = cls._client._get(API_URL.DEPLOY_GET_SERVICE_API, data)
    if "data" in rsp:
        checked = ServiceApiInfo._converter.check(rsp['data'])
        filtered = Deployment._filter_data(checked)
        return ServiceApiInfo(**filtered)
    return None

get_service_status(self)

获取服务状态

Returns:
  • Int – 服务状态:1-创建,2-调度中,3-调度成功,4-执行中,5-创建失败,6-挂起,7-取消,8-删除中,9- 已删除

Source code in deepwisdom/models/deployment.py
def get_service_status(self) -> Int:
    """获取服务状态

    Returns:
        Int: 服务状态:1-创建,2-调度中,3-调度成功,4-执行中,5-创建失败,6-挂起,7-取消,8-删除中,9- 已删除
    """
    svc = self.get_service(self.id)
    return svc.status

list_deployments(project_id, **kwargs) classmethod

获取服务部署列表

Parameters:
  • project_id (int) – 项目id

Returns:
  • List[DeploymentListMember] – 服务列表

Source code in deepwisdom/models/deployment.py
@classmethod
def list_deployments(cls, project_id: int, **kwargs) -> List["DeploymentListMember"]:
    """获取服务部署列表

    Args:
        project_id (int): 项目id

    Returns:
        List[DeploymentListMember]: 服务列表
    """
    data = {
        "project_id": project_id
    }
    server_data = cls._server_data(API_URL.DEPLOY_LIST_DEPLOYMENTS, data)
    # return [cls.get_deployment_detail(item["service_id"]) for item in server_data]
    return [DeploymentListMember(**item) for item in server_data]

rename_deployment(svc_id, svc_name) classmethod

重命名服务

Parameters:
  • svc_id (int) – 服务id

  • svc_name (str) – 新服务名称

Returns:
  • Deployment – 服务详情

Source code in deepwisdom/models/deployment.py
@classmethod
def rename_deployment(cls, svc_id: int, svc_name: str):
    """重命名服务

    Args:
        svc_id (int): 服务id
        svc_name (str): 新服务名称
    Returns:
        Deployment: 服务详情
    """
    data = {
        "svc_id": svc_id,
        "svc_name": svc_name
    }
    rsp = cls._client._patch(API_URL.DEPLOY_RENAME_DEPLOYMENT, data)
    if rsp['code'] == 200 and rsp['message'] == 'ok':
        return cls.get_deployment_detail(svc_id)
    return None

resident_deployment(svc_id, min_pod) classmethod

切换服务常驻状态

Parameters:
  • svc_id (int64) – 服务id

  • min_pod (int) – 最少pod数,大于0为服务常驻,等于0为非常驻

Returns:
  • Deployment – 服务详情

Source code in deepwisdom/models/deployment.py
@classmethod
def resident_deployment(cls, svc_id: int, min_pod: int):
    """切换服务常驻状态

    Args:
        svc_id (int64): 服务id
        min_pod (int): 最少pod数,大于0为服务常驻,等于0为非常驻
    Returns:
        Deployment: 服务详情
    """
    data = {
        "svc_id": svc_id,
        "min_pod": min_pod
    }
    rsp = cls._client._post(API_URL.DEPLOY_RESIDENT_DEPLOYMENT, data)
    if rsp['code'] == 200 and rsp['message'] == 'ok':
        return cls.get_deployment_detail(svc_id)
    return None

DeploymentListMember (APIObject) dataclass

DeploymentListMember(service_id: int = 0, service_name: str = '', min_pod: int = 0, model_name: str = '', trial_no: int = 0, trial_type: int = 0, recom_model: str = '', eval_metric: trafaret.base.Any = '', service_status: int = 0, service_create_time: str = '')

get_deployment_detail(self)

获取服务详情

Returns:
  • Deployment – 服务详情

Source code in deepwisdom/models/deployment.py
def get_deployment_detail(self) -> "Deployment":
    """获取服务详情
    Returns:
        Deployment: 服务详情
    """
    data = {
        "service_id": self.service_id
    }
    rsp = self._client._get(API_URL.DEPLOY_GET_SERVICE_DETAIL, data)
    if "data" in rsp:
        checked = Deployment._converter.check(rsp['data'])
        filtered = Deployment._filter_data(checked)
        return Deployment(**filtered)

ServiceApiInfo (APIObject) dataclass

ServiceApiInfo(http_type: str = '', token: str = '', url: str = '', post_example: trafaret.base.Any = '', data_example: trafaret.base.Any = '', response_example: trafaret.base.Any = None)